"""Границы трёх утверждений статьи — блоки 17-19.

ЗАЧЕМ ЭТОТ СКРИПТ. Разбор снаружи указал на три места, где статья сказала
шире, чем измерила. Все три можно не переписать словами, а перемерить, и
здесь они перемерены:

 1. «Реплика не повысится НИКОГДА» — это верно при значении
    `cluster-replica-validity-factor` по умолчанию. Настройка существует
    именно затем, чтобы это менять, и блок 17 показывает, что при нуле ответ
    другой.
 2. «Кворум выполнен, а переключения нет» — хорошо, но в проде такое надо
    уметь проверять ЗАРАНЕЕ, а не по факту. Для этого есть отдельная команда,
    и блок 18 показывает, что она отвечает по обе стороны от границы.
 3. «Узел данных про топологию не знает ничего» — слишком широко. Про
    авторитетное мнение службы он действительно не знает, а вот про свою пару
    репликации знает вполне; блок 19 печатает, что именно.

ЗАПУСК: python3 bench/valkey-cluster/boundaries.py
"""

from __future__ import annotations

import os
import re
import time

from common import (
    CLUSTER_NODE_TIMEOUT_MS,
    Cluster,
    cli,
    cluster_state,
    free_port,
    head,
    link_status,
    note,
    preamble,
    role_of,
    row,
)

SENTINEL_NAME = "cache"
PAD = 40


def replica_of(ports: list[int], master_port: int) -> int | None:
    raw = cli(ports[0], "cluster", "nodes")
    master_id = None
    for line in raw.splitlines():
        f = line.split()
        if f[1].startswith(f"127.0.0.1:{master_port}@"):
            master_id = f[0]
    if master_id is None:
        return None
    for line in raw.splitlines():
        f = line.split()
        if "slave" in f[2] and f[3] == master_id:
            return int(f[1].split("@")[0].split(":")[1])
    return None


def block17(cluster: Cluster) -> None:
    head(17, "ТО ЖЕ ОКНО, НО С ОТКЛЮЧЁННОЙ ПРОВЕРКОЙ СВЕЖЕСТИ")

    # Кластер поднимается с validity-factor 0 и роняется внутри того же окна,
    # что и в блоке 12: единственное отличие — одна настройка.
    ports = [free_port() for _ in range(6)]
    for p in ports:
        cluster.start_data(p, cluster=True, extra=["--cluster-replica-validity-factor", "0"])
    cluster.form_cluster_raw(ports, replicas=1)

    row("cluster-replica-validity-factor",
        cli(ports[0], "config", "get", "cluster-replica-validity-factor").split()[-1], pad=PAD)
    row("cluster-node-timeout", f"{CLUSTER_NODE_TIMEOUT_MS} мс", pad=PAD)
    print()

    deadline = time.time() + 60
    while time.time() < deadline and not all(cluster_state(p) == "ok" for p in ports):
        time.sleep(0.05)
    masters = [p for p in ports if role_of(p) == "master"]
    victim = masters[0]
    rep = replica_of(ports, victim)
    row("мастер, который упадёт", f"127.0.0.1:{victim}", pad=PAD)
    row("его реплика", f"127.0.0.1:{rep}", pad=PAD)
    row("master_link_status реплики", link_status(rep), pad=PAD)
    print()

    cluster.kill(victim)
    print(f"  t=0.00  мастер {victim} убит, ждём переключения 40 с")
    t0 = time.time()
    promoted = False
    last = None
    while time.time() - t0 < 40:
        state = cluster_state(next(p for p in ports if p != victim))
        role = role_of(rep)
        now = (state, role)
        if now != last:
            print(f"  t={time.time() - t0:5.2f}  состояние кластера {state}, роль реплики {role}")
            last = now
        if role == "master":
            promoted = True
            break
        time.sleep(0.2)

    print()
    row("реплика повысилась", "да" if promoted else "НЕТ, за 40 с", pad=PAD)

    note(
        """
        Тот же опыт, что в блоке 12, с единственным отличием в одной строке
        конфига — и другой исход. Значит, «никогда» из блока 12 относится не к
        Valkey вообще, а к значению `cluster-replica-validity-factor` по
        умолчанию: проверку свежести данных реплики можно отключить, и тогда
        она согласится повыситься, не завершив ни одной синхронизации.

        Это не совет так делать. Проверка стоит там не зря: реплика, никогда
        не синхронизировавшаяся с мастером, повышается ПУСТОЙ, и переключение
        такого рода лечит доступность ценой данных. Но знать, что поведение
        настраивается, надо — иначе блок 12 читается как неустранимое
        свойство, а он про настройку.
        """
    )
    for p in ports:
        cluster.kill(p)


def block18(cluster: Cluster) -> None:
    head(18, "КАК ПРОВЕРИТЬ КВОРУМ И БОЛЬШИНСТВО ЗАРАНЕЕ, А НЕ ПО ФАКТУ")

    master = free_port()
    replicas = [free_port(), free_port()]
    sentinels = [free_port() for _ in range(5)]
    cluster.start_data(master)
    for r in replicas:
        cluster.start_data(r, replicaof=master)
    for s in sentinels:
        cluster.start_sentinel(s, master, quorum=2, name=SENTINEL_NAME)
    time.sleep(5)

    watcher = sentinels[0]
    row("sentinel в наборе / кворум", "5 / 2", pad=PAD)
    row("живых sentinel", "5", pad=PAD)
    row("SENTINEL CKQUORUM cache",
        cli(watcher, "sentinel", "ckquorum", SENTINEL_NAME), pad=PAD)
    print()

    for s in sentinels[2:]:
        cluster.kill(s)
    time.sleep(8)
    row("живых sentinel", "2", pad=PAD)
    row("SENTINEL CKQUORUM cache",
        cli(watcher, "sentinel", "ckquorum", SENTINEL_NAME), pad=PAD)

    note(
        """
        Та же граница, что в блоке 13, но увиденная заранее и одной командой.
        Пока наблюдателей большинство, ответ утвердительный и в нём названы
        оба условия. Когда большинство потеряно, команда отвечает отказом — и
        отказывает она именно по второму условию, хотя кворума из настройки
        хватает.

        Практическая ценность в том, что это можно спрашивать постоянно, а не
        выяснять в момент аварии. Ответ на эту команду — готовый признак для
        наблюдения: он отвечает на вопрос «переключимся ли мы, если сейчас
        упадёт мастер», а не «всё ли хорошо прямо сейчас».
        """
    )
    for p in [master] + replicas + sentinels[:2]:
        cluster.kill(p)


def block19(cluster: Cluster) -> None:
    head(19, "ЧТО УЗЕЛ ПОД SENTINEL ВСЁ-ТАКИ ЗНАЕТ О ТОПОЛОГИИ")

    master = free_port()
    replicas = [free_port(), free_port()]
    sentinel = free_port()
    cluster.start_data(master)
    for r in replicas:
        cluster.start_data(r, replicaof=master)
    cluster.start_sentinel(sentinel, master, quorum=1, name=SENTINEL_NAME)
    time.sleep(4)

    info = cli(master, "info", "replication")
    print("  мастер про себя (INFO replication):")
    for line in info.splitlines():
        line = line.strip()
        if re.match(r"^(role|connected_slaves|slave\d+):", line):
            print("    " + line)
    print()
    print("  реплика про себя:")
    info_r = cli(replicas[0], "info", "replication")
    for line in info_r.splitlines():
        line = line.strip()
        if re.match(r"^(role|master_host|master_port|master_link_status):", line):
            print("    " + line)
    print()
    row("мастер: ROLE (первое поле)", cli(master, "role").splitlines()[0], pad=PAD)
    row("мастер: CLUSTER INFO", cli(master, "cluster", "info").splitlines()[0], pad=PAD)
    row("мастер: кто мастер по мнению службы",
        cli(master, "sentinel", "get-master-addr-by-name", SENTINEL_NAME).splitlines()[0], pad=PAD)

    note(
        """
        Уточнение к формулировке, которая в первой версии статьи была шире
        измеренного. Узел под sentinel знает не «ничего» — он знает свою пару
        репликации: мастер перечисляет подключённые реплики, реплика называет
        адрес мастера и состояние канала. Это настоящая топологическая
        информация, и наблюдению она вполне годится.

        Чего он не знает — так это АВТОРИТЕТНОГО ответа на вопрос «кто мастер
        этой службы сейчас». Своё мнение у него есть, и в момент переключения
        оно бывает устаревшим: ровно это и происходит в блоке 10, где
        вернувшийся узел искренне считает мастером себя. Отказ на
        `SENTINEL get-master-addr-by-name` в последней строке — не про
        отсутствие данных, а про то, что эта команда принадлежит другому
        набору и другой службе.
        """
    )


def main() -> None:
    preamble("три утверждения статьи, проверенные на границах")

    cluster = Cluster()
    try:
        block17(cluster)
        block18(cluster)
        block19(cluster)
    finally:
        cluster.stop_all()
    _ = os


if __name__ == "__main__":
    main()