"""Мастер в одном датацентре, реплика в другом: что теряется при переключении.

ЧТО ЗДЕСЬ МЕРЯЕТСЯ. Не «медленно ли работает репликация» — это и так
очевидно, — а сколько ПОДТВЕРЖДЁННЫХ клиенту записей исчезает, когда мастер
пропал и реплику подняли на его место. Величина важна тем, что её обычно
считают нулевой: клиент получил OK, значит записалось.

ПОЧЕМУ ЭТО НЕ ПРИДИРКА К REDIS. Асинхронная репликация — не дефект
реализации, а объявленный договор: мастер отвечает клиенту, не дожидаясь
реплик. Документация Redis говорит об этом прямо, и цитата стоит в статье.
Замер лишь переводит договор в число для конкретного расстояния.

ЗАПУСК: python3 bench/stretched-cache/replication.py
Нужен redis-server в PATH. Ничего не остаётся после выхода: оба процесса
поднимаются во временных каталогах и убиваются в finally.
"""

from __future__ import annotations

import os
import shutil
import socket
import subprocess
import sys
import tempfile
import time

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from link import DelayedLink  # noqa: E402

import redis  # noqa: E402

# Какой сервер запускать. По умолчанию тот, что в PATH; REDIS_SERVER_BIN и
# REDIS_CLI_BIN позволяют прогнать тот же замер на другой сборке — так снята
# таблица регрессии на Redis 8.10.1 и smoke-прогон на Valkey.
SERVER = os.environ.get("REDIS_SERVER_BIN", "redis-server")
CLI = os.environ.get("REDIS_CLI_BIN", "redis-cli")

# ОДНОСТОРОННЯЯ задержка. Круг — вдвое больше. Двадцать миллисекунд в одну
# сторону это типичное расстояние между датацентрами в пределах одной страны:
# Москва — Петербург около того. Величина выбрана не «чтобы было хуже», а
# как обычный случай, о котором и приходят просить.
ONE_WAY_MS = 20.0
WRITES = 200


def free_port() -> int:
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
    s.close()
    return port


def start_redis(port: int, dirname: str) -> subprocess.Popen:
    proc = subprocess.Popen(
        [
            SERVER,
            "--port", str(port),
            "--dir", dirname,
            "--save", "",
            "--appendonly", "no",
            "--daemonize", "no",
            "--logfile", os.path.join(dirname, "redis.log"),
        ]
    )
    deadline = time.time() + 10
    while time.time() < deadline:
        try:
            c = redis.Redis(port=port, socket_timeout=0.5)
            if c.ping():
                c.close()
                return proc
        except Exception:
            time.sleep(0.05)
    raise RuntimeError(f"redis на порту {port} не поднялся")


def wait_sync(client: redis.Redis, timeout: float = 15.0) -> None:
    """Дождаться, пока реплика доложит link_status:up и первую синхронизацию."""
    deadline = time.time() + timeout
    while time.time() < deadline:
        info = client.info("replication")
        if info.get("master_link_status") == "up":
            return
        time.sleep(0.1)
    raise RuntimeError("реплика не синхронизировалась")


def main() -> None:
    tmp_a = tempfile.mkdtemp(prefix="dc-a-")
    tmp_b = tempfile.mkdtemp(prefix="dc-b-")
    port_a, port_b = free_port(), free_port()
    proc_a = proc_b = None
    try:
        proc_a = start_redis(port_a, tmp_a)
        proc_b = start_redis(port_b, tmp_b)
        a = redis.Redis(port=port_a, decode_responses=True)
        b = redis.Redis(port=port_b, decode_responses=True)

        version = a.info("server")["redis_version"]
        print(f"Redis {version} · один хост, два процесса")
        print(f"канал между ДЦ: {ONE_WAY_MS:.0f} мс в одну сторону, "
              f"то есть круг {2 * ONE_WAY_MS:.0f} мс")
        print()

        # --- Блок 1: цена расстояния для самой репликации -------------------
        near = DelayedLink(port_a, delay_ms=0.0)
        b.replicaof("127.0.0.1", near.port)
        wait_sync(b)
        a.set("probe", "1")
        t0 = time.perf_counter()
        while b.get("probe") != "1":
            time.sleep(0.0005)
        near_lag = (time.perf_counter() - t0) * 1000

        b.execute_command("REPLICAOF", "NO", "ONE")
        b.flushall()
        far = DelayedLink(port_a, delay_ms=ONE_WAY_MS)
        b.replicaof("127.0.0.1", far.port)
        wait_sync(b)
        a.set("probe2", "1")
        t0 = time.perf_counter()
        while b.get("probe2") != "1":
            time.sleep(0.0005)
        far_lag = (time.perf_counter() - t0) * 1000

        print("1. ЗАПИСЬ ДОХОДИТ ДО РЕПЛИКИ НЕ СРАЗУ, И РАССТОЯНИЕ ЭТО ВИДНО")
        print("-" * 66)
        print(f"  реплика в том же ДЦ: запись видна через, мс   {near_lag:.1f}")
        print(f"  реплика в другом ДЦ: запись видна через, мс   {far_lag:.1f}")
        print(f"  разница, мс                                   {far_lag - near_lag:.1f}")
        print()
        print("  Клиент к этому моменту давно получил OK: мастер отвечает,")
        print("  не дожидаясь реплики. Отставание само по себе не беда —")
        print("  бедой оно становится в следующем блоке.")
        print()

        # --- Блок 2: подтверждённые записи, которых не стало ----------------
        a.flushall()
        time.sleep(0.3)
        acked = 0
        for i in range(WRITES):
            a.set(f"key:{i}", i)
            acked += 1
        # Разрыв ровно в тот момент, когда все записи подтверждены клиенту.
        far.cut()
        # Реплику поднимают на место мастера: это и есть переключение.
        b.execute_command("REPLICAOF", "NO", "ONE")
        time.sleep(0.2)
        survived = sum(1 for i in range(WRITES) if b.get(f"key:{i}") is not None)
        lost = acked - survived

        # Та же связка, но разрыв не сразу после записей, а через секунду.
        # ЗАЧЕМ ВТОРАЯ ПОЛОВИНА: без неё число из первой читается как
        # «теряется почти всё», а это неверно. Теряется то, что не успело
        # улететь, и вопрос только в том, сколько записей помещается в один
        # круг. Двести записей по 0,15 мс помещаются целиком — отсюда и 99,5%.
        far.cut()
        b.execute_command("REPLICAOF", "NO", "ONE")
        b.flushall()
        far3 = DelayedLink(port_a, delay_ms=ONE_WAY_MS)
        b.replicaof("127.0.0.1", far3.port)
        wait_sync(b)
        a.flushall()
        time.sleep(0.3)
        for i in range(WRITES):
            a.set(f"calm:{i}", i)
        time.sleep(1.0)          # тишина: репликация догнала
        far3.cut()
        b.execute_command("REPLICAOF", "NO", "ONE")
        time.sleep(0.2)
        calm_survived = sum(1 for i in range(WRITES) if b.get(f"calm:{i}") is not None)

        print("2. ПЕРЕКЛЮЧЕНИЕ НА РЕПЛИКУ ТЕРЯЕТ ПОДТВЕРЖДЁННЫЕ ЗАПИСИ")
        print("-" * 66)
        print(f"  записей подтверждено клиенту                  {acked}")
        print("  разрыв сразу после записи:")
        print(f"    оказалось на реплике                        {survived}")
        print(f"    потеряно                                    {lost} ({100 * lost / acked:.1f}%)")
        print("  разрыв через секунду после записи:")
        print(f"    оказалось на реплике                        {calm_survived}")
        print(f"    потеряно                                    {acked - calm_survived}"
              f" ({100 * (acked - calm_survived) / acked:.1f}%)")
        print()
        print("  Читать надо обе строки сразу. Теряется не «почти всё» и не")
        print("  «ничего»: теряется то, что в момент разрыва ещё не доехало")
        print("  до реплики. Залп из двухсот записей уходит быстрее, чем")
        print("  летит, — отсюда первая строка. Через секунду тишины")
        print("  репликация догнала — отсюда вторая.")
        print()
        print("  ЧЕМ ИЗМЕРЯЕТСЯ ЭТО ОКНО, ЗДЕСЬ НЕ ВИДНО, и соблазн ответить")
        print("  «кругом» велик. В первой версии статьи так и было написано,")
        print("  и это неверно: круг здесь совпал с потерей случайно. Блок 9")
        print("  меняет один лишь ТЕМП записи на том же канале и получает")
        print("  другие потери — смотреть надо на разницу смещений в")
        print("  INFO replication, а не на круг.")
        print()
        print("  Ни одна из потерянных записей об ошибке не сообщила — на все")
        print("  клиент получил OK. Это не сбой репликации, а её договор:")
        print("  мастер не ждёт реплику.")
        print()

        # --- Блок 3: WAIT покупает не то, за что его принимают --------------
        far2 = DelayedLink(port_a, delay_ms=ONE_WAY_MS)
        b.replicaof("127.0.0.1", far2.port)
        wait_sync(b)
        a.flushall()
        time.sleep(0.3)

        t0 = time.perf_counter()
        for i in range(20):
            a.set(f"plain:{i}", i)
        plain_ms = (time.perf_counter() - t0) * 1000 / 20

        t0 = time.perf_counter()
        acked_by_replica = 0
        for i in range(20):
            a.set(f"waited:{i}", i)
            acked_by_replica += a.execute_command("WAIT", 1, 1000)
        wait_ms = (time.perf_counter() - t0) * 1000 / 20

        print("3. WAIT ДЕЛАЕТ ЗАПИСЬ ДОРОЖЕ, НО НЕ ДЕЛАЕТ ЕЁ ТРАНЗАКЦИЕЙ")
        print("-" * 66)
        print(f"  обычная запись, мс                            {plain_ms:.2f}")
        print(f"  запись с WAIT 1, мс                           {wait_ms:.2f}")
        print(f"  во сколько раз дороже                         {wait_ms / plain_ms:.1f}")
        print(f"  реплик подтвердило (из 20 записей по 1)       {acked_by_replica}")
        print()
        print("  Цена — круг между датацентрами на КАЖДУЮ запись. А купленное")
        print("  свойство слабее ожидаемого: WAIT сообщает, сколько реплик")
        print("  подтвердило, но не отменяет запись, если их меньше. К моменту")
        print("  ответа WAIT запись на мастере уже применена и уже видна")
        print("  читателям — откатить её нечем.")
        print()
    finally:
        for proc in (proc_b, proc_a):
            if proc is not None:
                proc.terminate()
                try:
                    proc.wait(timeout=5)
                except subprocess.TimeoutExpired:
                    proc.kill()
        shutil.rmtree(tmp_a, ignore_errors=True)
        shutil.rmtree(tmp_b, ignore_errors=True)


if __name__ == "__main__":
    main()