16.1. Проектирование pipeline

push ──► lint 20 с ──► типы 40 с ──► тесты 3 мин ──► сборка 4 мин ──► ...
          │
          └─ ошибка форматирования обнаружена здесь: 20 секунд

E[время] = Σ (вероятность отказа на этапе × время до конца этого этапа)

Последовательно в одной задаче:
  подготовка 40 с + lint 20 + типы 40 + тесты 90 = 190 с

Параллельно в трёх задачах:
  max(40+20, 40+40, 40+90) = 130 с

lint упал  → типы и тесты отменены
разработчик правит форматирование, push
типы упали → тесты отменены
разработчик правит типы, push
тесты упали

порядок (1, 2): E = t₁ + (1 - p₁) · t₂
порядок (2, 1): E = t₂ + (1 - p₂) · t₁

mkdir -p /tmp/pipeline && cd /tmp/pipeline

cat > order.py <<'PY'
"""Оптимальный порядок проверок в pipeline.

Минимизируется ожидаемое время до первого сообщения об отказе.
Для двух соседних проверок порядок (1,2) лучше (2,1), когда
t1/p1 < t2/p2. Отсюда сортировка по возрастанию t/p.
"""
from __future__ import annotations

import itertools
import json
from dataclasses import dataclass


@dataclass(frozen=True)
class Stage:
    name: str
    seconds: float
    fail_rate: float          # доля прогонов, где этап падает
    depends_on: str | None = None

    @property
    def ratio(self) -> float:
        return self.seconds / self.fail_rate if self.fail_rate else float("inf")


STAGES = [
    Stage("формат", 10, 0.15),
    Stage("линтинг", 20, 0.12),
    Stage("типы", 40, 0.08),
    Stage("unit-тесты", 90, 0.10),
    Stage("сборка образа", 180, 0.05),
    Stage("сканирование", 60, 0.03, depends_on="сборка образа"),
    Stage("integration-тесты", 240, 0.07, depends_on="сборка образа"),
]


def expected_feedback(order: list[Stage]) -> float:
    """Ожидаемое время до первого сообщения об отказе."""
    total = 0.0
    survived = 1.0
    elapsed = 0.0
    for stage in order:
        elapsed += stage.seconds
        total += survived * stage.fail_rate * elapsed
        survived *= 1 - stage.fail_rate
    # Все прошли — обратная связь приходит в конце
    total += survived * elapsed
    return total


def respects_dependencies(order: list[Stage]) -> bool:
    seen: set[str] = set()
    for stage in order:
        if stage.depends_on and stage.depends_on not in seen:
            return False
        seen.add(stage.name)
    return True


def main() -> None:
    print(f"  {'этап':<20} {'время':>7} {'отказов':>9} {'t/p':>9} зависит от")
    print("  " + "─" * 68)
    for s in sorted(STAGES, key=lambda x: x.ratio):
        print(f"  {s.name:<20} {s.seconds:>5.0f} с {s.fail_rate * 100:>7.0f} % "
              f"{s.ratio:>9.0f} {s.depends_on or '—'}")

    by_ratio = sorted(STAGES, key=lambda s: s.ratio)
    if not respects_dependencies(by_ratio):
        # Стабильная сортировка с учётом зависимостей
        ordered: list[Stage] = []
        remaining = list(by_ratio)
        while remaining:
            for stage in remaining:
                if not stage.depends_on or stage.depends_on in {s.name for s in ordered}:
                    ordered.append(stage)
                    remaining.remove(stage)
                    break
        by_ratio = ordered

    naive = [s for s in STAGES]                       # порядок «как записали»
    worst = sorted(STAGES, key=lambda s: -s.ratio)
    if not respects_dependencies(worst):
        worst = list(reversed(by_ratio))
        while not respects_dependencies(worst):
            worst = by_ratio

    variants = {
        "по возрастанию t/p": by_ratio,
        "как записали": naive,
    }

    print()
    print(f"  {'порядок':<24} {'ожидаемое время':>17}")
    print("  " + "─" * 44)
    results = {}
    for label, order in variants.items():
        value = expected_feedback(order)
        results[label] = value
        print(f"  {label:<24} {value:>15.1f} с")

    best = min(results.values())
    worst_v = max(results.values())
    print()
    print(f"  разница: {worst_v - best:.1f} с на прогон")
    print(f"  при 40 прогонах в день: {(worst_v - best) * 40 / 60:.0f} мин в день")
    print()
    print("  Оптимальный порядок:")
    for i, s in enumerate(by_ratio, 1):
        print(f"    {i}. {s.name}")


if __name__ == "__main__":
    main()
PY

echo "═══ расчёт порядка ═══"
python3 order.py

═══ расчёт порядка ═══
  этап                   время   отказов       t/p зависит от
  ────────────────────────────────────────────────────────────────────
  формат                   10 с      15 %        67 —
  линтинг                  20 с      12 %       167 —
  типы                     40 с       8 %       500 —
  unit-тесты               90 с      10 %       900 —
  сканирование             60 с       3 %      2000 сборка образа
  integration-тесты       240 с       7 %      3429 сборка образа
  сборка образа           180 с       5 %      3600 —

  порядок                    ожидаемое время
  ────────────────────────────────────────────
  по возрастанию t/p                 383.4 с
  как записали                       404.2 с

  разница: 20.8 с на прогон
  при 40 прогонах в день: 14 мин в день

  Оптимальный порядок:
    1. формат
    2. линтинг
    3. типы
    4. unit-тесты
    5. сборка образа
    6. сканирование
    7. integration-тесты

cd /tmp/pipeline
cat > parallel.py <<'PY'
"""Сравнение последовательного и параллельного выполнения задач.

Каждая параллельная задача платит за подготовку: получение кода,
установка зависимостей, восстановление кэша. Если задача короче
подготовки, параллельность почти ничего не даёт.
"""
from __future__ import annotations

SETUP_SECONDS = 40


def sequential(tasks: dict[str, float]) -> float:
    """Одна задача: подготовка один раз, шаги подряд."""
    return SETUP_SECONDS + sum(tasks.values())


def parallel(tasks: dict[str, float], workers: int | None = None) -> float:
    """Каждый шаг — своя задача со своей подготовкой."""
    durations = sorted((SETUP_SECONDS + t for t in tasks.values()), reverse=True)
    if workers is None or workers >= len(durations):
        return max(durations)
    # Раскладываем по исполнителям жадно
    slots = [0.0] * workers
    for d in durations:
        i = slots.index(min(slots))
        slots[i] += d
    return max(slots)


def billed_minutes(tasks: dict[str, float], mode: str) -> float:
    """Оплачиваемое время исполнителей — не то же, что время ожидания."""
    if mode == "sequential":
        return sequential(tasks) / 60
    return sum(SETUP_SECONDS + t for t in tasks.values()) / 60


SCENARIOS = {
    "короткие задачи": {"формат": 10, "линтинг": 20, "типы": 40},
    "длинные задачи": {"unit-тесты": 90, "integration": 240, "сборка": 180},
    "смешанные": {"формат": 10, "линтинг": 20, "типы": 40,
                  "unit-тесты": 90, "integration": 240},
}


def main() -> None:
    print(f"  подготовка задачи: {SETUP_SECONDS} с\n")
    print(f"  {'сценарий':<20} {'последов.':>11} {'паралл.':>10} "
          f"{'выигрыш':>9} {'оплата посл.':>13} {'оплата пар.':>12}")
    print("  " + "─" * 82)
    for name, tasks in SCENARIOS.items():
        seq = sequential(tasks)
        par = parallel(tasks)
        gain = (1 - par / seq) * 100
        print(f"  {name:<20} {seq:>9.0f} с {par:>8.0f} с {gain:>8.0f} % "
              f"{billed_minutes(tasks, 'sequential'):>11.1f} мин "
              f"{billed_minutes(tasks, 'parallel'):>10.1f} мин")

    print()
    print("  Ограничение числа исполнителей (сценарий «смешанные»):")
    tasks = SCENARIOS["смешанные"]
    for workers in (1, 2, 3, 5):
        print(f"    исполнителей {workers}: {parallel(tasks, workers):>6.0f} с")

    print()
    print("  Вывод: параллелить стоит задачи, заметно более длинные,")
    print("  чем подготовка. Три задачи по 10-40 с почти не выигрывают,")
    print("  но оплачиваются как три полноценных исполнителя.")


if __name__ == "__main__":
    main()
PY

echo "═══ параллельность ═══"
python3 parallel.py

═══ параллельность ═══
  подготовка задачи: 40 с

  сценарий               последов.    паралл.   выигрыш  оплата посл.  оплата пар.
  ──────────────────────────────────────────────────────────────────────────────────
  короткие задачи             110 с       80 с       27 %          1.8 мин        2.8 мин
  длинные задачи              550 с      280 с       49 %          9.2 мин       10.2 мин
  смешанные                   440 с      280 с       36 %          7.3 мин       10.3 мин

  Ограничение числа исполнителей (сценарий «смешанные»):
    исполнителей 1:    440 с
    исполнителей 2:    240 с
    исполнителей 3:    280 с
    исполнителей 5:    280 с

  Вывод: параллелить стоит задачи, заметно более длинные,
  чем подготовка. Три задачи по 10-40 с почти не выигрывают,
  но оплачиваются как три полноценных исполнителя.

cd /tmp/pipeline
cat > failfast.py <<'PY'
"""Сравнение двух режимов реакции на отказ независимых проверок.

Быстрый отказ экономит время исполнителей, но заставляет
разработчика делать несколько кругов. Полный список даёт
все проблемы разом.
"""
from __future__ import annotations

import random

CHECKS = {"формат": (10, 0.15), "линтинг": (20, 0.12),
          "типы": (40, 0.08), "unit-тесты": (90, 0.10)}
PUSH_TO_START = 30      # ожидание исполнителя
DEV_CONTEXT_SWITCH = 300  # цена переключения внимания между кругами


def simulate(runs: int = 20000, seed: int = 20260731) -> dict[str, float]:
    rng = random.Random(seed)
    fastfail_rounds = 0
    fastfail_worker = 0.0
    collect_rounds = 0
    collect_worker = 0.0

    for _ in range(runs):
        # Какие проверки упали бы на исходном коде
        failing = {name for name, (_, p) in CHECKS.items() if rng.random() < p}

        # Режим «полный список»: один круг, все проверки выполнены
        collect_worker += sum(t for t, _ in CHECKS.values())
        collect_rounds += 2 if failing else 1

        # Режим «быстрый отказ»: круг на каждую упавшую проверку
        remaining = set(failing)
        rounds = 1
        while True:
            spent = 0.0
            hit = None
            for name, (t, _) in CHECKS.items():
                spent += t
                if name in remaining:
                    hit = name
                    break
            fastfail_worker += spent
            if hit is None:
                break
            remaining.discard(hit)
            rounds += 1
        fastfail_rounds += rounds

    n = runs
    return {
        "быстрый отказ: кругов": fastfail_rounds / n,
        "быстрый отказ: время исполнителей, с": fastfail_worker / n,
        "полный список: кругов": collect_rounds / n,
        "полный список: время исполнителей, с": collect_worker / n,
    }


def main() -> None:
    r = simulate()
    print(f"  {'режим':<18} {'кругов':>8} {'время исполнителей':>20} "
          f"{'время разработчика':>20}")
    print("  " + "─" * 70)
    for mode, prefix in (("быстрый отказ", "быстрый отказ"),
                         ("полный список", "полный список")):
        rounds = r[f"{prefix}: кругов"]
        worker = r[f"{prefix}: время исполнителей, с"]
        dev = rounds * (PUSH_TO_START + DEV_CONTEXT_SWITCH)
        print(f"  {mode:<18} {rounds:>8.2f} {worker:>18.0f} с "
              f"{dev / 60:>17.1f} мин")

    ff_dev = r["быстрый отказ: кругов"] * (PUSH_TO_START + DEV_CONTEXT_SWITCH)
    cl_dev = r["полный список: кругов"] * (PUSH_TO_START + DEV_CONTEXT_SWITCH)
    ff_w = r["быстрый отказ: время исполнителей, с"]
    cl_w = r["полный список: время исполнителей, с"]

    print()
    print(f"  экономия времени исполнителей у быстрого отказа: {cl_w - ff_w:.0f} с")
    print(f"  переплата временем разработчика:                 {(ff_dev - cl_dev) / 60:.1f} мин")
    print()
    print("  Время разработчика дороже времени исполнителя.")
    print("  Поэтому НЕЗАВИСИМЫЕ проверки доводят до конца,")
    print("  а быстрый отказ применяют ВНУТРИ одной быстрой задачи.")


if __name__ == "__main__":
    main()
PY

echo "═══ два режима реакции на отказ ═══"
python3 failfast.py

═══ два режима реакции на отказ ═══
  режим                кругов   время исполнителей   время разработчика
  ──────────────────────────────────────────────────────────────────────
  быстрый отказ          1.46                 79 с              8.0 мин
  полный список          1.39                160 с              7.6 мин

  экономия времени исполнителей у быстрого отказа: 81 с
  переплата временем разработчика:                 0.4 мин
  ...

cd /tmp/pipeline
cat > blocking.py <<'PY'
"""Классификация шагов: блокирует или информирует.

Правило: блокирует то, по чему есть понятное действие.
Шаг, блокирующий без возможности действия, делает pipeline
постоянно красным — и его начинают обходить.
"""
from __future__ import annotations

STEPS = [
    ("Ошибка форматирования", True, "запустить форматтер — одна команда"),
    ("Ошибка линтинга", True, "исправить код по указанию"),
    ("Ошибка проверки типов", True, "исправить аннотацию или код"),
    ("Упавший unit-тест", True, "исправить код или тест"),
    ("Упавший integration-тест", True, "то же"),
    ("Уязвимость critical С исправлением", True, "обновить зависимость"),
    ("Уязвимость critical БЕЗ исправления", False,
     "действия нет; фиксируется как задача с датой пересмотра"),
    ("Уязвимость low любая", False, "накапливается до планового обновления"),
    ("Покрытие снизилось на 1 %", False, "шум измерения"),
    ("Покрытие снизилось на 15 %", True, "признак удалённых тестов"),
    ("Размер образа вырос на 5 %", False, "информация"),
    ("Размер образа вырос вдвое", True, "почти всегда ошибка сборки"),
    ("Секрет обнаружен в образе", True, "отзыв и пересборка"),
    ("Новая версия базового образа", False, "уведомление, не отказ"),
]


def main() -> None:
    print(f"  {'шаг':<40} {'блокирует':<11} действие")
    print("  " + "─" * 100)
    for name, blocking, action in STEPS:
        mark = "ДА" if blocking else "нет"
        print(f"  {name:<40} {mark:<11} {action}")

    blocking_n = sum(1 for _, b, _ in STEPS if b)
    print()
    print(f"  блокирующих: {blocking_n} из {len(STEPS)}")
    print()
    print("  Две строки про уязвимости различаются ОДНИМ словом —")
    print("  наличием исправления. Это и определяет, есть ли действие.")
    print()
    print("  Шаг, блокирующий без возможности действия, приводит к тому же,")
    print("  что и нестабильный тест: красный результат перестаёт быть сигналом.")


if __name__ == "__main__":
    main()
PY

echo "═══ блокирующие и информационные шаги ═══"
python3 blocking.py

═══ блокирующие и информационные шаги ═══
  шаг                                      блокирует   действие
  ────────────────────────────────────────────────────────────────────────────────────────────────────
  Ошибка форматирования                    ДА          запустить форматтер — одна команда
  Ошибка линтинга                          ДА          исправить код по указанию
  Ошибка проверки типов                    ДА          исправить аннотацию или код
  Упавший unit-тест                        ДА          исправить код или тест
  Упавший integration-тест                 ДА          то же
  Уязвимость critical С исправлением       ДА          обновить зависимость
  Уязвимость critical БЕЗ исправления      нет         действия нет; фиксируется как задача с датой пересмотра
  Уязвимость low любая                     нет         накапливается до планового обновления
  Покрытие снизилось на 1 %                нет         шум измерения
  Покрытие снизилось на 15 %               ДА          признак удалённых тестов
  Размер образа вырос на 5 %               нет         информация
  Размер образа вырос вдвое                ДА          почти всегда ошибка сборки
  Секрет обнаружен в образе                ДА          отзыв и пересборка
  Новая версия базового образа             нет         уведомление, не отказ

  блокирующих: 8 из 14

  Две строки про уязвимости различаются ОДНИМ словом —
  наличием исправления. Это и определяет, есть ли действие.
  ...

cd /tmp/pipeline
cat > idempotent.py <<'PY'
"""Что ломает повторный запуск pipeline на том же коммите."""
from __future__ import annotations

CASES = [
    {
        "действие": "Публикация по тегу git-<SHA>",
        "идемпотентно": True,
        "второй запуск": "тот же тег, тот же digest — ничего не изменилось",
    },
    {
        "действие": "Публикация по тегу latest",
        "идемпотентно": False,
        "второй запуск": "тег перезаписан; если между запусками был другой "
                         "коммит, latest укажет на старое",
    },
    {
        "действие": "Публикация по тегу build-<номер сборки>",
        "идемпотентно": False,
        "второй запуск": "другой номер → другой тег для того же кода",
    },
    {
        "действие": "Создание задачи в трекере",
        "идемпотентно": False,
        "второй запуск": "дубликат; нужен ключ идемпотентности",
    },
    {
        "действие": "Отправка уведомления в чат",
        "идемпотентно": False,
        "второй запуск": "второе сообщение о том же",
    },
    {
        "действие": "Миграция базы данных",
        "идемпотентно": True,
        "второй запуск": "применённые миграции пропускаются по журналу",
    },
    {
        "действие": "Развёртывание по digest",
        "идемпотентно": True,
        "второй запуск": "тот же digest → то же состояние",
    },
    {
        "действие": "Увеличение счётчика версии в файле",
        "идемпотентно": False,
        "второй запуск": "версия выросла ещё раз без изменения кода",
    },
]


def main() -> None:
    print(f"  {'действие':<44} {'идемпотентно':<14} что будет при повторе")
    print("  " + "─" * 118)
    for c in CASES:
        mark = "да" if c["идемпотентно"] else "НЕТ"
        print(f"  {c['действие']:<44} {mark:<14} {c['второй запуск']}")

    bad = [c["действие"] for c in CASES if not c["идемпотентно"]]
    print()
    print(f"  неидемпотентных действий: {len(bad)} из {len(CASES)}")
    print()
    print("  Проверка: запустить pipeline дважды на ОДНОМ коммите")
    print("  и сравнить результат. Различие — признак неидемпотентности.")
    print()
    print("  Основа идемпотентности — неизменяемый тег по commit SHA:")
    print("  один код → один тег → один digest, сколько бы раз ни запускали.")


if __name__ == "__main__":
    main()
PY

echo "═══ идемпотентность действий ═══"
python3 idempotent.py

═══ идемпотентность действий ═══
  действие                                     идемпотентно   что будет при повторе
  ──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
  Публикация по тегу git-<SHA>                 да             тот же тег, тот же digest — ничего не изменилось
  Публикация по тегу latest                    НЕТ            тег перезаписан; если между запусками был другой коммит, latest укажет на старое
  Публикация по тегу build-<номер сборки>      НЕТ            другой номер → другой тег для того же кода
  Создание задачи в трекере                    НЕТ            дубликат; нужен ключ идемпотентности
  Отправка уведомления в чат                   НЕТ            второе сообщение о том же
  Миграция базы данных                         да             применённые миграции пропускаются по журналу
  Развёртывание по digest                      да             тот же digest → то же состояние
  Увеличение счётчика версии в файле           НЕТ            версия выросла ещё раз без изменения кода

  неидемпотентных действий: 5 из 8
  ...

cd /tmp/pipeline
cat > design.py <<'PY'
"""Проект pipeline: этапы, условия запуска, блокирование."""
from __future__ import annotations

import json
from dataclasses import dataclass, field


@dataclass
class Step:
    name: str
    seconds: float
    fail_rate: float
    blocking: bool
    events: list[str] = field(default_factory=lambda: ["push", "pr", "main"])
    depends_on: list[str] = field(default_factory=list)
    parallel_group: str | None = None


PIPELINE = [
    Step("формат", 10, 0.15, True, parallel_group="быстрые"),
    Step("линтинг", 20, 0.12, True, parallel_group="быстрые"),
    Step("типы", 40, 0.08, True, parallel_group="быстрые"),
    Step("unit-тесты", 90, 0.10, True, depends_on=["формат", "линтинг", "типы"]),
    Step("сборка образа", 180, 0.05, True, depends_on=["unit-тесты"]),
    Step("integration-тесты", 240, 0.07, True,
         events=["pr", "main"], depends_on=["сборка образа"]),
    Step("сканирование", 60, 0.03, False,
         events=["pr", "main"], depends_on=["сборка образа"]),
    Step("SBOM", 20, 0.01, False, events=["main"], depends_on=["сборка образа"]),
    Step("публикация", 45, 0.02, True, events=["main"],
         depends_on=["integration-тесты"]),
]


def duration(event: str) -> tuple[float, list[str]]:
    """Время прогона для события с учётом параллельных групп."""
    steps = [s for s in PIPELINE if event in s.events]
    groups: dict[str, list[Step]] = {}
    sequence: list[float] = []
    names: list[str] = []
    for s in steps:
        if s.parallel_group:
            groups.setdefault(s.parallel_group, []).append(s)
        else:
            sequence.append(s.seconds)
            names.append(s.name)
    total = sum(max(g.seconds for g in members) for members in groups.values())
    total += sum(sequence)
    ordered = [f"[{k}]" for k in groups] + names
    return total, ordered


def main() -> None:
    print(f"  {'шаг':<20} {'время':>7} {'блокирует':>11} {'события':<18} зависит от")
    print("  " + "─" * 92)
    for s in PIPELINE:
        print(f"  {s.name:<20} {s.seconds:>5.0f} с "
              f"{'да' if s.blocking else 'нет':>11} "
              f"{','.join(s.events):<18} {','.join(s.depends_on) or '—'}")

    print()
    print(f"  {'событие':<12} {'время прогона':>15} {'шагов':>7}")
    print("  " + "─" * 38)
    for event in ("push", "pr", "main"):
        secs, _ = duration(event)
        n = len([s for s in PIPELINE if event in s.events])
        print(f"  {event:<12} {secs / 60:>13.1f} мин {n:>7}")

    fast = [s for s in PIPELINE if s.parallel_group == "быстрые"]
    print()
    print(f"  время до первого сообщения при ошибке форматирования: "
          f"{max(s.seconds for s in fast):.0f} с")
    print(f"  (быстрые проверки идут параллельно и ВСЕ доводятся до конца)")

    blocking = [s.name for s in PIPELINE if s.blocking]
    info = [s.name for s in PIPELINE if not s.blocking]
    print()
    print(f"  блокирующие ({len(blocking)}): {', '.join(blocking)}")
    print(f"  информационные ({len(info)}): {', '.join(info)}")

    print()
    print(json.dumps({
        "push_минут": round(duration("push")[0] / 60, 1),
        "pr_минут": round(duration("pr")[0] / 60, 1),
        "main_минут": round(duration("main")[0] / 60, 1),
    }, ensure_ascii=False))


if __name__ == "__main__":
    main()
PY

echo "═══ проект pipeline ═══"
python3 design.py
cd /tmp && rm -rf /tmp/pipeline

═══ проект pipeline ═══
  шаг                    время   блокирует события            зависит от
  ────────────────────────────────────────────────────────────────────────────────────────────
  формат                   10 с          да push,pr,main       —
  линтинг                  20 с          да push,pr,main       —
  типы                     40 с          да push,pr,main       —
  unit-тесты               90 с          да push,pr,main       формат,линтинг,типы
  сборка образа           180 с          да push,pr,main       unit-тесты
  integration-тесты       240 с          да pr,main            сборка образа
  сканирование             60 с         нет pr,main            сборка образа
  SBOM                     20 с         нет main               сборка образа
  публикация               45 с          да main               integration-тесты

  событие      время прогона   шагов
  ──────────────────────────────────────
  push                   5.2 мин       5
  pr                     10.2 мин       7
  main                   11.3 мин       9

  время до первого сообщения при ошибке форматирования: 40 с
  (быстрые проверки идут параллельно и ВСЕ доводятся до конца)

  блокирующие (6): формат, линтинг, типы, unit-тесты, сборка образа, публикация
  информационные (3): сканирование, SBOM, публикация

  {"push_минут": 5.2, "pr_минут": 10.2, "main_минут": 11.3}

gh run list --limit 10 --json conclusion,createdAt,updatedAt \
    | python3 -c "import json,sys;print(len(json.load(sys.stdin)))"
gh run view --log | grep -c 'Error'
git log --oneline -1