"""Одни и те же опыты на трёх сборках: блок 15.

ЗАЧЕМ. Все числа статьи сняты на Redis 7.0.15. Ветка старая: к сентябрю 2026
актуальная линия Redis Open Source — 8.10.x, а Valkey ушёл в 9.1. Разбор
снаружи указал на это прямо: выводы сверены с текущей документацией, но
измерены на одной старой сборке, и «совпало однажды» не делает свойство
неизменным. Кластерная часть Redis между 7 и 8 менялась заметно, так что
проверять надо именно её.

ЧТО ЗДЕСЬ ПРОВЕРЯЕТСЯ. Не производительность — она на разных сборках разная по
скучным причинам. Проверяются УТВЕРЖДЕНИЯ статьи, каждое в том виде, в каком
его можно подтвердить или опровергнуть:

  1. мастер отвечает клиенту, не дожидаясь реплики;
  2. видимость записи на реплике отстаёт примерно на перелёт;
  3. WAIT 1 стоит примерно круг;
  4. разрыв сразу после залпа записей теряет подтверждённые записи;
  5. расклад B (узлы поровну, мастера 2:1) отказывает в половине с
     большинством мастеров;
  6. `cluster-allow-reads-when-down` решает судьбу чтений в отказавшем
     кластере.

7.0.15 остаётся опубликованной основой — на ней сняты все остальные блоки.
Этот блок отвечает на другой вопрос: не разъехалось ли поведение.

ЗАПУСК: python3 bench/stretched-cache/versions.py
Пути к сборкам — в DE_REDIS_BUILDS, через запятую, вида «имя=каталог/bin».
По умолчанию берутся те, что нашлись в системе и в /usr/local.
"""

from __future__ import annotations

import os
import socket
import statistics
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 cluster as clu  # noqa: E402
import redis  # noqa: E402

ONE_WAY_MS = 20.0
WRITES = 20
BURST = 200
PAD = 30

DEFAULT_BUILDS = [
    ("Redis 7.0.15", "/usr/bin"),
    ("Redis 8.10.1", "/usr/local/redis8101/bin"),
    ("Valkey 9.1.2", "/usr/local/valkey912/bin"),
]


def builds() -> list[tuple[str, str, str]]:
    """(подпись, путь к серверу, путь к cli) для каждой найденной сборки."""
    raw = os.environ.get("DE_REDIS_BUILDS")
    pairs = []
    if raw:
        for item in raw.split(","):
            name, _, path = item.partition("=")
            pairs.append((name.strip(), path.strip()))
    else:
        pairs = DEFAULT_BUILDS
    out = []
    for name, path in pairs:
        for server, cli in (("redis-server", "redis-cli"), ("valkey-server", "valkey-cli")):
            s, c = os.path.join(path, server), os.path.join(path, cli)
            if os.path.exists(s) and os.path.exists(c):
                out.append((name, s, c))
                break
    return out


def free_port(limit: int = 55535) -> int:
    for _ in range(500):
        s = socket.socket()
        s.bind(("127.0.0.1", 0))
        port = s.getsockname()[1]
        s.close()
        if port < limit:
            return port
    raise RuntimeError("не нашлось свободного порта")


def start(server: str, port: int, dirname: str, extra: list[str] | None = None):
    proc = subprocess.Popen([
        server, "--port", str(port), "--dir", dirname,
        "--save", "", "--appendonly", "no",
        "--logfile", os.path.join(dirname, "server.log"),
    ] + (extra or []))
    deadline = time.time() + 20
    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"сервер на порту {port} не поднялся")


def stop(*procs) -> None:
    for proc in procs:
        if proc is None:
            continue
        proc.terminate()
        try:
            proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            proc.kill()


def median_ms(f, n: int) -> float:
    samples = []
    for i in range(n):
        t0 = time.perf_counter()
        f(i)
        samples.append((time.perf_counter() - t0) * 1000.0)
    return statistics.median(samples)


def wait_sync(client: redis.Redis, timeout: float = 25.0) -> None:
    deadline = time.time() + timeout
    while time.time() < deadline:
        if client.info("replication").get("master_link_status") == "up":
            return
        time.sleep(0.05)
    raise RuntimeError("реплика не синхронизировалась")


def pair_checks(server: str) -> dict[str, str]:
    """Пункты 1-4: всё, для чего хватает мастера и реплики за каналом."""
    out: dict[str, str] = {}
    dir_a, dir_b = tempfile.mkdtemp(prefix="v-a-"), tempfile.mkdtemp(prefix="v-b-")
    port_a, port_b = free_port(), free_port()
    proc_a = proc_b = None
    link = None
    try:
        proc_a, proc_b = start(server, port_a, dir_a), start(server, port_b, dir_b)
        a = redis.Redis(port=port_a, decode_responses=True)
        b = redis.Redis(port=port_b, decode_responses=True)
        info = a.info("server")
        # У Valkey поле redis_version — это версия СОВМЕСТИМОСТИ (7.2.4), а не
        # своя. Своя лежит в valkey_version, и печатать надо её, иначе колонка
        # врёт ровно про то, ради чего этот блок и поставлен.
        out["версия"] = info.get("valkey_version") or info.get("redis_version", "?")

        link = DelayedLink(port_a, delay_ms=ONE_WAY_MS)
        b.replicaof("127.0.0.1", link.port)
        wait_sync(b)
        time.sleep(0.6)
        a.set("warm", "1")
        a.wait(1, 5000)

        # 1. Мастер не ждёт реплику: обычная запись стоит доли миллисекунды,
        #    хотя реплика за двадцатью миллисекундами.
        plain = median_ms(lambda i: a.set(f"p:{i}", i), WRITES)
        out["обычная запись, мс"] = f"{plain:.2f}"
        out["мастер ждёт реплику"] = "нет" if plain < ONE_WAY_MS / 2 else "ДА"

        # 2. Видимость записи на реплике.
        a.set("vis", "x0")
        time.sleep(0.5)
        t0 = time.perf_counter()
        a.set("vis", "x1")
        while time.perf_counter() - t0 < 5:
            if b.get("vis") == "x1":
                break
            time.sleep(0.0005)
        out["видна на реплике через, мс"] = f"{(time.perf_counter() - t0) * 1000:.1f}"

        # 3. WAIT.
        def with_wait(i: int) -> None:
            a.set(f"w:{i}", i)
            a.wait(1, 5000)

        waited = median_ms(with_wait, WRITES)
        out["WAIT 1, мс"] = f"{waited:.2f}"
        out["WAIT / круг"] = f"{waited / (2 * ONE_WAY_MS):.2f}"

        # 4. Потери при разрыве сразу после залпа.
        a.flushall()
        time.sleep(0.3)
        for i in range(BURST):
            a.set(f"burst:{i}", i)
        link.cut()
        time.sleep(1.0)
        b.replicaof("no", "one")
        on_replica = sum(1 for i in range(BURST) if b.get(f"burst:{i}") is not None)
        out[f"потеряно из {BURST} при разрыве"] = str(BURST - on_replica)
    finally:
        if link is not None:
            link.cut()
        stop(proc_a, proc_b)
    return out


def cluster_checks(server: str, cli: str) -> dict[str, str]:
    """Пункты 5-6: расклад B и судьба чтений в отказавшем кластере."""
    out: dict[str, str] = {}
    clu.SERVER, clu.CLI = server, cli

    for reads_when_down in ("no", "yes"):
        c = clu.Cluster()
        try:
            ports = c.start(6)
            # Настройка ставится на все узлы уже поднятого кластера: она
            # изменяемая, и отдельный кластер ради неё поднимать не нужно.
            for p in ports:
                clu.conn(p).config_set("cluster-allow-reads-when-down", reads_when_down)
            pairing = clu.pairing_a(ports)
            c.form(pairing)
            dc1, dc2 = clu.dcs_b(ports, pairing)

            cc = redis.RedisCluster(host="127.0.0.1", port=ports[0], decode_responses=True)
            keys = [f"k:{i}" for i in range(20)]
            for k in keys:
                cc.set(k, "v")
            cc.close()

            c.stop(dc2)
            time.sleep(clu.SETTLE_S)

            states = [clu.state_of(p) for p in dc1]
            answers = [clu.can_read(p, "dc:probe") for p in dc1]
            served = sum(1 for p in dc1 for k in keys
                         if clu.can_read(p, k) == "ответил")
            if reads_when_down == "no":
                out["расклад B: состояние половины с большинством"] = \
                    "/".join(sorted(set(states)))
                out["расклад B: ответов на GET"] = \
                    "нет" if all(a == "CLUSTERDOWN" for a in answers) else "есть"
            out[f"reads-when-down {reads_when_down}: отдано ключей из 60"] = str(served)
        finally:
            c.destroy()
    return out


def main() -> None:
    found = builds()
    if not found:
        print("не найдено ни одной сборки; укажите DE_REDIS_BUILDS")
        return

    print("одни и те же опыты на разных сборках; канал между ДЦ "
          f"{ONE_WAY_MS:.0f} мс в одну сторону")
    print("проверяются утверждения статьи, а не производительность")
    print()
    print("15. ТЕ ЖЕ УТВЕРЖДЕНИЯ НА ТЕКУЩИХ СБОРКАХ")
    print("-" * 66)

    results: list[tuple[str, dict[str, str]]] = []
    for name, server, cli in found:
        data = pair_checks(server)
        data.update(cluster_checks(server, cli))
        results.append((name, data))

    keys: list[str] = []
    for _, data in results:
        for k in data:
            if k not in keys:
                keys.append(k)

    width = max(len(k) for k in keys) + 2
    header = " " * (width + 2) + "".join(f"{name:>16}" for name, _ in results)
    print(header)
    for k in keys:
        line = f"  {k:<{width}}" + "".join(f"{data.get(k, '—'):>16}" for _, data in results)
        print(line)

    print()
    print("  Как это читать. Колонки не обязаны совпадать по числам: сборки")
    print("  разные, машина одна, и абсолютные миллисекунды тут ничего не")
    print("  значат. Значение имеют строки, где написаны СВОЙСТВА: ждёт ли")
    print("  мастер реплику, во сколько кругов обходится WAIT, теряются ли")
    print("  подтверждённые записи, отказывает ли половина с большинством")
    print("  мастеров и меняет ли reads-when-down судьбу чтений.")
    print()
    print("  Именно на этих строках держатся выводы статьи, и именно их")
    print("  разбор просил не считать неизменными только потому, что они")
    print("  однажды совпали с документацией.")


if __name__ == "__main__":
    main()