"""Почему отказавший кластер не отвечает и на чтение — и что это меняет.

ЗАЧЕМ ЭТОТ ЗАМЕР ПОЯВИЛСЯ. В первой версии статьи стояло утверждение:
документация `cluster-require-full-coverage` говорит про ЗАПИСИ, а замер
показал `CLUSTERDOWN` и на обычный `GET`, — значит, «наблюдаемое поведение
шире написанного». Утверждение оказалось неверным: у Redis есть отдельная
настройка `cluster-allow-reads-when-down`, по умолчанию `no`, и именно она
отвечает за чтения на узле отказавшего кластера. То есть поведение не шире
документации — оно описано, просто в другом месте.

Переписывать вывод словами было бы недостаточно: получилось бы одно
утверждение без замера вместо другого утверждения без замера. Поэтому здесь
тот же расклад, что в блоке 5, прогоняется ДВАЖДЫ — с настройкой `no` и с
настройкой `yes`, — и разница видна на одном и том же `GET`.

ЧТО ЭТО НЕ ОТМЕНЯЕТ. Главный результат статьи — «отказали обе половины» —
держится не на этой настройке, а на кворуме и покрытии слотов. Настройка
меняет только то, что видит клиент на чтении в отказавшей половине. Замер
ниже показывает и это.

ЗАПУСК: python3 bench/stretched-cache/readsdown.py
Нужны redis-server и redis-cli в PATH. Занимает около минуты.
"""

from __future__ import annotations

import os
import socket
import subprocess
import tempfile
import time

import redis

# Какой сервер запускать. По умолчанию тот, что в 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")

NODE_TIMEOUT_MS = 2000
SETTLE_S = 8.0

_PIDS: dict[int, int] = {}


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


def start_node(port: int, dirname: str, allow_reads: str) -> subprocess.Popen:
    proc = subprocess.Popen(
        [
            SERVER,
            "--port", str(port),
            "--dir", dirname,
            "--cluster-enabled", "yes",
            "--cluster-config-file", f"nodes-{port}.conf",
            "--cluster-node-timeout", str(NODE_TIMEOUT_MS),
            # Единственное, что различается между двумя прогонами.
            "--cluster-allow-reads-when-down", allow_reads,
            "--save", "",
            "--appendonly", "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()
                _PIDS[port] = proc.pid
                return proc
        except Exception:
            time.sleep(0.05)
    raise RuntimeError(f"узел {port} не поднялся")


def state_of(port: int) -> str:
    try:
        c = redis.Redis(port=port, socket_timeout=1.0, decode_responses=True)
        return c.cluster("INFO")["cluster_state"]
    except Exception:
        return "недоступен"


def probe(port: int, key: str, write: bool) -> str:
    """Что получает клиент: значение, перенаправление, отказ или молчание."""
    try:
        c = redis.Redis(port=port, socket_timeout=1.5, decode_responses=True)
        if write:
            c.set(key, "v")
            return "записал"
        value = c.get(key)
        return "ответил" if value is not None else "ответил (пусто)"
    except redis.exceptions.ClusterDownError:
        return "CLUSTERDOWN"
    except (redis.exceptions.MovedError, redis.exceptions.AskError):
        return "перенаправил"
    except redis.exceptions.ResponseError as exc:
        return str(exc).split()[0]
    except Exception:
        return "таймаут"


# Ключи раскладываются по слотам хешем, поэтому одного ключа мало: он
# случайно попадёт на какой-то один узел, и по нему нельзя сказать, что
# происходит с остальным пространством ключей. Двадцать ключей покрывают
# все три слотовых диапазона.
KEYS = [f"probe:{i}" for i in range(20)]


def probe_all(port: int) -> dict[str, int]:
    """Сколько ключей из KEYS узел отдал, перенаправил и по скольким отказал."""
    counts: dict[str, int] = {}
    for k in KEYS:
        answer = probe(port, k, write=False)
        counts[answer] = counts.get(answer, 0) + 1
    return counts


def topology(ports: list[int]) -> dict[int, str]:
    c = redis.Redis(port=ports[0], socket_timeout=2.0, decode_responses=True)
    raw = c.execute_command("CLUSTER", "NODES")
    by_id: dict[str, int] = {}
    rows = []
    for line in raw.strip().splitlines():
        f = line.split()
        node_id, addr, flags, master_id = f[0], f[1], f[2], f[3]
        port = int(addr.split("@")[0].split(":")[1])
        by_id[node_id] = port
        rows.append((port, flags, master_id))
    return {
        port: ("мастер" if "master" in flags else f"реплика {by_id.get(master_id, '?')}")
        for port, flags, master_id in rows
    }


def roles(ports: list[int]) -> dict[int, str]:
    out = {}
    for p in ports:
        try:
            c = redis.Redis(port=p, socket_timeout=1.0, decode_responses=True)
            out[p] = c.info("replication")["role"]
        except Exception:
            out[p] = "?"
    return out


def run_once(allow_reads: str) -> list[tuple[int, str, str, str]]:
    """Поднять кластер, разложить по расклад B, оборвать канал, опросить ДЦ-1.

    Возвращает строки (порт, роль, состояние, ответ на GET) для той половины,
    у которой большинство мастеров, — той самой, про которую и шёл спор.
    """
    tmp = tempfile.mkdtemp(prefix=f"readsdown-{allow_reads}-")
    ports = [free_port() for _ in range(6)]
    procs = []
    try:
        for p in ports:
            d = os.path.join(tmp, str(p))
            os.makedirs(d, exist_ok=True)
            procs.append(start_node(p, d, allow_reads))

        args = [CLI, "--cluster", "create"]
        args += [f"127.0.0.1:{p}" for p in ports]
        args += ["--cluster-replicas", "1", "--cluster-yes"]
        subprocess.run(args, check=True, capture_output=True, timeout=60)
        time.sleep(3)

        cc = redis.RedisCluster(host="127.0.0.1", port=ports[0], decode_responses=True)
        for k in KEYS:
            cc.set(k, "value")
        cc.close()

        topo = topology(ports)
        masters = [p for p, v in topo.items() if v == "мастер"]
        replicas = [p for p in ports if p not in masters]

        # Расклад B из блока 5: узлы поровну, мастера поделены 2 и 1, и
        # реплика третьего мастера уезжает вместе с ним.
        dc1 = [masters[0], masters[1], replicas[2]]
        dc2 = [masters[2], replicas[0], replicas[1]]

        for p in dc2:
            subprocess.run(["kill", "-STOP", str(_PIDS[p])], check=False)
        time.sleep(SETTLE_S)

        r = roles(dc1)
        rows = []
        for p in dc1:
            counts = probe_all(p)
            summary = ", ".join(f"{v} {k}" for k, v in sorted(counts.items()))
            rows.append((p, r.get(p, "?"), state_of(p), summary))
        # Заодно спросим запись — чтобы было видно, что настройка про чтения
        # и только про них.
        write_answer = probe(dc1[0], KEYS[0], write=True)
        rows.append((0, "запись", "", write_answer))

        for p in dc2:
            subprocess.run(["kill", "-CONT", str(_PIDS[p])], check=False)
        return rows
    finally:
        for proc in procs:
            proc.terminate()
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                proc.kill()


def main() -> None:
    print("Redis · тот же расклад, что в блоке 5: узлы поровну, мастера 2 и 1")
    print(f"cluster-node-timeout {NODE_TIMEOUT_MS} мс; после разрыва {SETTLE_S:.0f} с на выборы")
    print("опрашивается половина С БОЛЬШИНСТВОМ мастеров — та, про которую спор")
    print()

    print("8. ЧТЕНИЯ В ОТКАЗАВШЕМ КЛАСТЕРЕ РЕШАЕТ ОТДЕЛЬНАЯ НАСТРОЙКА")
    print("-" * 66)
    served: dict[str, int] = {}
    for allow in ("no", "yes"):
        rows = run_once(allow)
        # Сколько ключей из двадцати выжившая половина реально отдала.
        # Считается по выводу, а не пишется руками: какие именно мастера
        # окажутся в ДЦ-1, зависит от того, как кластер расставил роли, и
        # число от прогона к прогону разное.
        served[allow] = sum(
            int(part.split()[0])
            for _, role, _, answer in rows
            if role != "запись"
            for part in answer.split(", ")
            if part.endswith("ответил")
        )
        print(f"  cluster-allow-reads-when-down {allow}"
              f"{'  (по умолчанию)' if allow == 'no' else ''}:")
        for port, role, state, answer in rows:
            if role == "запись":
                print(f"    SET на том же узле                        -> {answer}")
            else:
                print(f"    узел {port}: {role:<8} состояние {state:<8} 20 ключей -> {answer}")
        print()

    print("  Обе строки сняты на одном и том же раскладе и одном и том же")
    print("  разрыве. Различается одна настройка.")
    print()
    print("  Отсюда поправка к первой версии статьи. Утверждение «наблюдаемое")
    print("  поведение шире документации» было неверным: отказ на чтении не")
    print("  побочный эффект cluster-require-full-coverage, а отдельно")
    print("  описанное поведение cluster-allow-reads-when-down со значением")
    print("  no по умолчанию. Цепочка такая: неполное покрытие слотов ->")
    print("  кластер помечен отказавшим -> узел отказавшего кластера не")
    print("  обслуживает чтения, пока не сказано обратное.")
    print()
    print("  И чего настройка НЕ меняет — это важнее самой поправки.")
    print()
    print("  Со значением yes узел отвечает только по ТЕМ ключам, чьи слоты")
    print("  остались за ним: остальные он перенаправляет на мастер, который")
    print("  сейчас за оборванным каналом, и туда клиент не доедет.")
    print()
    print(f"  Ключей отдано из двадцати: с no — {served['no']}, с yes — "
          f"{served['yes']}.")
    print("  Второе число — не двадцать и от прогона к прогону разное: оно")
    print("  зависит от того, сколько слотов досталось двум оставшимся")
    print("  мастерам. Треть пространства ключей — слоты мастера за обрывом —")
    print("  недоступна при любой настройке.")
    print()
    print("  Записи отказаны в обоих случаях. То есть yes не возвращает кэш:")
    print("  он превращает «не работает ничего» в «работает часть чтений на")
    print("  части пространства ключей». Главный вывод статьи — при ровном")
    print("  делении кэша не остаётся ни в одном ДЦ — от настройки не зависит.")


if __name__ == "__main__":
    main()