10.4. Database migrations

# Кажется удобным
@asynccontextmanager
async def lifespan(app):
    run_migrations()          # ← опасно
    yield

SELECT pg_advisory_lock(hashtext('schema_migrations'));
-- миграции
SELECT pg_advisory_unlock(hashtext('schema_migrations'));

def connect_with_retry(dsn: str, attempts: int = 30) -> Connection:
    delay = 0.25
    for _ in range(attempts):
        try:
            return psycopg.connect(dsn, connect_timeout=3)
        except psycopg.OperationalError:
            time.sleep(delay)
            delay = min(delay * 2, 3.0)      # экспоненциальная задержка
    raise RuntimeError("база недоступна")

1. expand:   добавить новую колонку, оставив старую
2. deploy:   код пишет в обе, читает из новой
3. backfill: заполнить новую колонку из старой
4. deploy:   код использует только новую
5. contract: удалить старую колонку

services:
  db-test:
    image: postgres:17-alpine
    environment:
      POSTGRES_PASSWORD: test
      POSTGRES_DB: testdb
    tmpfs:
      - /var/lib/postgresql/data:size=512m     # в памяти, исчезает
    command:
      - postgres
      - -c
      - fsync=off                    # ТОЛЬКО для тестов
      - -c
      - full_page_writes=off
      - -c
      - synchronous_commit=off

BEGIN;
CREATE TABLE a (...);
ALTER TABLE b ADD COLUMN c int;
-- ошибка здесь откатит обе операции
COMMIT;

mkdir -p /tmp/migr && cd /tmp/migr

cat > racy.py <<'PY'
"""Миграция БЕЗ блокировки — воспроизводит гонку."""
import os
import sys
import time

import psycopg

DSN = os.environ["DSN"]
NAME = os.environ.get("REPLICA", "?")


def connect_with_retry(attempts: int = 40):
    delay = 0.25
    for _ in range(attempts):
        try:
            return psycopg.connect(DSN, connect_timeout=3)
        except psycopg.OperationalError:
            time.sleep(delay)
            delay = min(delay * 2, 3.0)
    raise RuntimeError("база недоступна")


def main() -> int:
    conn = connect_with_retry()
    conn.autocommit = True
    try:
        with conn.cursor() as cur:
            # Намеренно БЕЗ IF NOT EXISTS и без блокировки
            cur.execute("CREATE TABLE items (id serial PRIMARY KEY, name text)")
        print(f"{NAME}: таблица создана", flush=True)
        return 0
    except psycopg.errors.DuplicateTable:
        print(f"{NAME}: ОШИБКА — таблица уже существует", flush=True)
        return 1
    except Exception as exc:
        print(f"{NAME}: ОШИБКА — {type(exc).__name__}: {exc}", flush=True)
        return 1
    finally:
        conn.close()


if __name__ == "__main__":
    sys.exit(main())
PY

cat > compose.yaml <<'EOF'
name: migr

services:
  db:
    image: postgres:17-alpine
    environment:
      POSTGRES_PASSWORD: secret
      POSTGRES_DB: appdb
    tmpfs:
      - /var/lib/postgresql/data:size=512m
    command:
      - postgres
      - -c
      - fsync=off
      - -c
      - synchronous_commit=off
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres -d appdb"]
      interval: 2s
      timeout: 3s
      retries: 20
      start_period: 15s
      start_interval: 1s

  racy:
    image: python:3.13-slim
    volumes:
      - ./racy.py:/racy.py:ro
    environment:
      DSN: "postgresql://postgres:secret@db:5432/appdb"
      REPLICA: "${REPLICA:-r}"
    command: ["sh", "-c", "pip install -q 'psycopg[binary]==3.3.4' && python -u /racy.py"]
    depends_on:
      db:
        condition: service_healthy
    deploy:
      replicas: 3
    restart: "no"
EOF

docker compose down -v > /dev/null 2>&1
docker compose up -d db > /dev/null 2>&1
for _ in $(seq 60); do
    [ "$(docker compose ps --format '{{.Health}}' db)" = "healthy" ] && break
    sleep 1
done

echo "═══ три реплики выполняют миграцию одновременно ═══"
docker compose up racy 2>&1 | grep -E 'таблица создана|ОШИБКА|exited' | sed 's/^/  /'

═══ три реплики выполняют миграцию одновременно ═══
  migr-racy-1  | racy-1: таблица создана
  migr-racy-2  | racy-2: ОШИБКА — таблица уже существует
  migr-racy-3  | racy-3: ОШИБКА — таблица уже существует
  migr-racy-1 exited with code 0
  migr-racy-2 exited with code 1
  migr-racy-3 exited with code 1

cd /tmp/migr
cat > idempotent.py <<'PY'
"""Идемпотентная миграция: повтор безвреден."""
import os
import sys
import time

import psycopg

DSN = os.environ["DSN"]
NAME = os.environ.get("REPLICA", "?")


def connect_with_retry(attempts: int = 40):
    delay = 0.25
    for _ in range(attempts):
        try:
            return psycopg.connect(DSN, connect_timeout=3)
        except psycopg.OperationalError:
            time.sleep(delay)
            delay = min(delay * 2, 3.0)
    raise RuntimeError("база недоступна")


def main() -> int:
    with connect_with_retry() as conn, conn.cursor() as cur:
        cur.execute("CREATE TABLE IF NOT EXISTS items (id serial PRIMARY KEY, name text)")
        cur.execute("CREATE INDEX IF NOT EXISTS items_name_idx ON items (name)")
        conn.commit()
    print(f"{NAME}: миграция применена", flush=True)
    return 0


if __name__ == "__main__":
    sys.exit(main())
PY

python3 - <<'PY'
import pathlib
p = pathlib.Path("compose.yaml")
t = p.read_text().replace("./racy.py:/racy.py:ro", "./idempotent.py:/m.py:ro") \
                 .replace("python -u /racy.py", "python -u /m.py")
p.write_text(t)
PY

docker compose down > /dev/null 2>&1
docker compose up -d db > /dev/null 2>&1
for _ in $(seq 60); do
    [ "$(docker compose ps --format '{{.Health}}' db)" = "healthy" ] && break
    sleep 1
done

echo "═══ три реплики, идемпотентная миграция ═══"
docker compose up racy 2>&1 | grep -E 'применена|ОШИБКА|exited' | sed 's/^/  /'

═══ три реплики, идемпотентная миграция ═══
  migr-racy-1  | racy-1: миграция применена
  migr-racy-2  | racy-2: миграция применена
  migr-racy-3  | racy-3: миграция применена
  migr-racy-1 exited with code 0
  migr-racy-2 exited with code 0
  migr-racy-3 exited with code 0

cd /tmp/migr
cat > locked.py <<'PY'
"""Миграция под консультативной блокировкой: выполняется ровно один раз."""
import os
import sys
import time

import psycopg

DSN = os.environ["DSN"]
NAME = os.environ.get("REPLICA", "?")
LOCK_KEY = 987654321          # произвольный, но одинаковый у всех реплик
TARGET_VERSION = 2


def connect_with_retry(attempts: int = 40):
    delay = 0.25
    for _ in range(attempts):
        try:
            return psycopg.connect(DSN, connect_timeout=3)
        except psycopg.OperationalError:
            time.sleep(delay)
            delay = min(delay * 2, 3.0)
    raise RuntimeError("база недоступна")


MIGRATIONS = {
    1: [
        "CREATE TABLE IF NOT EXISTS items (id serial PRIMARY KEY, name text)",
        "CREATE INDEX IF NOT EXISTS items_name_idx ON items (name)",
    ],
    2: [
        "ALTER TABLE items ADD COLUMN IF NOT EXISTS created_at timestamptz DEFAULT now()",
    ],
}


def main() -> int:
    with connect_with_retry() as conn:
        with conn.cursor() as cur:
            # Блокировка освобождается автоматически при закрытии соединения
            cur.execute("SELECT pg_advisory_lock(%s)", (LOCK_KEY,))
            print(f"{NAME}: блокировка получена", flush=True)

            cur.execute("""
                CREATE TABLE IF NOT EXISTS schema_version (
                    version integer PRIMARY KEY,
                    applied_at timestamptz NOT NULL DEFAULT now()
                )
            """)
            cur.execute("SELECT coalesce(max(version), 0) FROM schema_version")
            current = cur.fetchone()[0]

            applied = 0
            for version in sorted(MIGRATIONS):
                if version <= current:
                    continue
                for stmt in MIGRATIONS[version]:
                    cur.execute(stmt)
                cur.execute("INSERT INTO schema_version (version) VALUES (%s)", (version,))
                applied += 1
                print(f"{NAME}: применена миграция {version}", flush=True)

            conn.commit()
            if applied == 0:
                print(f"{NAME}: нечего применять, схема версии {current}", flush=True)
            cur.execute("SELECT pg_advisory_unlock(%s)", (LOCK_KEY,))
    return 0


if __name__ == "__main__":
    sys.exit(main())
PY

python3 - <<'PY'
import pathlib
p = pathlib.Path("compose.yaml")
p.write_text(p.read_text().replace("./idempotent.py:/m.py:ro", "./locked.py:/m.py:ro"))
PY

docker compose down > /dev/null 2>&1
docker compose up -d db > /dev/null 2>&1
for _ in $(seq 60); do
    [ "$(docker compose ps --format '{{.Health}}' db)" = "healthy" ] && break
    sleep 1
done

echo "═══ три реплики под блокировкой ═══"
docker compose up racy 2>&1 | grep -E 'блокировка|применена|нечего|exited' | sed 's/^/  /'

echo "═══ итоговая схема ═══"
docker compose exec -T db psql -U postgres -d appdb -c '\d items' 2>/dev/null | head -8 | sed 's/^/  /'
docker compose exec -T db psql -U postgres -d appdb -tAc \
    'SELECT version, applied_at FROM schema_version ORDER BY version' | sed 's/^/  версия: /'

═══ три реплики под блокировкой ═══
  migr-racy-1  | racy-1: блокировка получена
  migr-racy-1  | racy-1: применена миграция 1
  migr-racy-1  | racy-1: применена миграция 2
  migr-racy-2  | racy-2: блокировка получена
  migr-racy-2  | racy-2: нечего применять, схема версии 2
  migr-racy-3  | racy-3: блокировка получена
  migr-racy-3  | racy-3: нечего применять, схема версии 2
  migr-racy-1 exited with code 0
  migr-racy-2 exited with code 0
  migr-racy-3 exited with code 0
═══ итоговая схема ═══
                                 Table "public.items"
     Column   |           Type           | Nullable |      Default      
    ----------+--------------------------+----------+-------------------
     id       | integer                  | not null | nextval(...)
     name     | text                     |          | 
     created_at | timestamp with time zone |        | now()
  версия: 1|2026-07-31 12:04:11.882+00
  версия: 2|2026-07-31 12:04:11.891+00

cd /tmp/migr
echo "═══ повторный запуск ═══"
docker compose up racy 2>&1 | grep -E 'применена|нечего' | sed 's/^/  /'

echo "═══ версия схемы не изменилась ═══"
docker compose exec -T db psql -U postgres -d appdb -tAc \
    'SELECT count(*) FROM schema_version' | xargs printf '  записей в schema_version: %s\n'

═══ повторный запуск ═══
  migr-racy-1  | racy-1: нечего применять, схема версии 2
  migr-racy-2  | racy-2: нечего применять, схема версии 2
  migr-racy-3  | racy-3: нечего применять, схема версии 2
═══ версия схемы не изменилась ═══
  записей в schema_version: 2

cd /tmp/migr
cat > failing.py <<'PY'
"""Миграция, падающая на середине. Проверяет откат DDL."""
import os
import sys

import psycopg

DSN = os.environ["DSN"]

with psycopg.connect(DSN) as conn, conn.cursor() as cur:
    cur.execute("SELECT count(*) FROM information_schema.tables WHERE table_name LIKE 'part_%'")
    print(f"  таблиц part_* до миграции: {cur.fetchone()[0]}", flush=True)

try:
    with psycopg.connect(DSN) as conn, conn.cursor() as cur:
        cur.execute("CREATE TABLE part_one (id int)")
        print("  создана part_one", flush=True)
        cur.execute("CREATE TABLE part_two (id int)")
        print("  создана part_two", flush=True)
        cur.execute("СИНТАКСИЧЕСКАЯ ОШИБКА")          # падение
        conn.commit()
except Exception as exc:
    print(f"  миграция упала: {type(exc).__name__}", flush=True)

with psycopg.connect(DSN) as conn, conn.cursor() as cur:
    cur.execute("SELECT count(*) FROM information_schema.tables WHERE table_name LIKE 'part_%'")
    n = cur.fetchone()[0]
    print(f"  таблиц part_* после падения: {n}", flush=True)
    sys.exit(0 if n == 0 else 1)
PY

docker compose run --rm -T \
    -v "$PWD/failing.py:/f.py:ro" \
    -e DSN="postgresql://postgres:secret@db:5432/appdb" \
    racy sh -c "pip install -q 'psycopg[binary]==3.3.4' && python -u /f.py" 2>/dev/null

  таблиц part_* до миграции: 0
  создана part_one
  создана part_two
  миграция упала: SyntaxError
  таблиц part_* после падения: 0

cd /tmp/migr
cat > expand.py <<'PY'
"""Шаг 1 (expand): добавить новую колонку, не трогая старую."""
import os

import psycopg

with psycopg.connect(os.environ["DSN"]) as conn, conn.cursor() as cur:
    cur.execute("ALTER TABLE items ADD COLUMN IF NOT EXISTS title text")
    conn.commit()
    print("  expand: колонка title добавлена, name сохранена", flush=True)
    cur.execute("""
        SELECT column_name FROM information_schema.columns
        WHERE table_name = 'items' AND column_name IN ('name', 'title')
        ORDER BY column_name
    """)
    print(f"  колонки: {[r[0] for r in cur.fetchall()]}", flush=True)
PY

cat > backfill.py <<'PY'
"""Шаг 3 (backfill): заполнить новую колонку из старой, порциями."""
import os

import psycopg

with psycopg.connect(os.environ["DSN"]) as conn, conn.cursor() as cur:
    cur.execute("INSERT INTO items (name) SELECT 'элемент-' || g FROM generate_series(1, 500) g")
    conn.commit()
    total = 0
    while True:
        # Порциями, чтобы не держать длинную транзакцию на большой таблице
        cur.execute("""
            UPDATE items SET title = name
            WHERE id IN (SELECT id FROM items WHERE title IS NULL LIMIT 100)
        """)
        conn.commit()
        if cur.rowcount == 0:
            break
        total += cur.rowcount
    print(f"  backfill: заполнено строк {total}", flush=True)
    cur.execute("SELECT count(*) FROM items WHERE title IS NULL")
    print(f"  осталось незаполненных: {cur.fetchone()[0]}", flush=True)
PY

cat > contract.py <<'PY'
"""Шаг 5 (contract): удалить старую колонку. ВЫПОЛНЯЕТСЯ ВРУЧНУЮ."""
import os

import psycopg

with psycopg.connect(os.environ["DSN"]) as conn, conn.cursor() as cur:
    cur.execute("SELECT count(*) FROM items WHERE title IS NULL")
    unfilled = cur.fetchone()[0]
    if unfilled:
        print(f"  contract ОТМЕНЁН: {unfilled} строк не заполнено", flush=True)
        raise SystemExit(1)
    cur.execute("ALTER TABLE items DROP COLUMN IF EXISTS name")
    conn.commit()
    print("  contract: колонка name удалена", flush=True)
PY

run() {
    docker compose run --rm -T \
        -v "$PWD/$1:/s.py:ro" \
        -e DSN="postgresql://postgres:secret@db:5432/appdb" \
        racy sh -c "pip install -q 'psycopg[binary]==3.3.4' && python -u /s.py" 2>/dev/null
}

echo "═══ шаг 1: expand ═══"
run expand.py
echo "═══ шаг 3: backfill ═══"
run backfill.py
echo "═══ шаг 5: contract (только вручную, после проверки) ═══"
run contract.py

═══ шаг 1: expand ═══
  expand: колонка title добавлена, name сохранена
  колонки: ['name', 'title']
═══ шаг 3: backfill ═══
  backfill: заполнено строк 500
  осталось незаполненных: 0
═══ шаг 5: contract (только вручную, после проверки) ═══
  contract: колонка name удалена

cd /tmp/migr
cat > compose.bench.yaml <<'EOF'
name: migr-bench

services:
  db-disk:
    image: postgres:17-alpine
    environment:
      POSTGRES_PASSWORD: test
      POSTGRES_DB: testdb
    volumes:
      - bench-data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres -d testdb"]
      interval: 1s
      retries: 60
      start_period: 20s

  db-tmpfs:
    image: postgres:17-alpine
    environment:
      POSTGRES_PASSWORD: test
      POSTGRES_DB: testdb
    tmpfs:
      - /var/lib/postgresql/data:size=512m
    command:
      - postgres
      - -c
      - fsync=off
      - -c
      - full_page_writes=off
      - -c
      - synchronous_commit=off
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres -d testdb"]
      interval: 1s
      retries: 60
      start_period: 20s

volumes:
  bench-data:
EOF

bench() {
    local svc="$1"
    docker compose -f compose.bench.yaml up -d "$svc" > /dev/null 2>&1
    for _ in $(seq 90); do
        [ "$(docker compose -f compose.bench.yaml ps --format '{{.Health}}' "$svc")" = "healthy" ] && break
        sleep 1
    done
    local s e
    s="$(date +%s.%N)"
    docker compose -f compose.bench.yaml exec -T "$svc" psql -U postgres -d testdb -q <<'SQL' > /dev/null 2>&1
CREATE TABLE bench (id serial PRIMARY KEY, payload text);
INSERT INTO bench (payload) SELECT md5(g::text) FROM generate_series(1, 20000) g;
CREATE INDEX bench_payload_idx ON bench (payload);
SQL
    e="$(date +%s.%N)"
    printf '  %-9s %.2f c\n' "$svc" "$(awk -v a="$s" -v b="$e" 'BEGIN{print b-a}')"
}

echo "═══ 20 000 вставок плюс индекс ═══"
bench db-disk
bench db-tmpfs
docker compose -f compose.bench.yaml down -v > /dev/null 2>&1

═══ 20 000 вставок плюс индекс ═══
  db-disk   1.84 c
  db-tmpfs  0.61 c

cd /tmp/migr && docker compose down -v > /dev/null 2>&1
cd /tmp && rm -rf /tmp/migr

mkdir -p /tmp/mg && cd /tmp/mg
cat > compose.yaml <<'EOF'
name: mg
services:
  db:
    image: postgres:17-alpine
    environment:
      POSTGRES_PASSWORD: x
      POSTGRES_DB: appdb
    tmpfs:
      - /var/lib/postgresql/data:size=256m
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres -d appdb"]
      interval: 2s
      retries: 20
      start_period: 15s
  migrate:
    image: postgres:17-alpine
    command:
      - sh
      - -c
      - 'PGPASSWORD=x psql -h db -U postgres -d appdb -c "CREATE TABLE IF NOT EXISTS t (id int)" && echo МИГРАЦИЯ ОК'
    depends_on:
      db:
        condition: service_healthy
EOF
docker compose up --abort-on-container-exit --exit-code-from migrate
docker compose down -v
cd /tmp && rm -rf /tmp/mg