"""Готов к работе и готов к отказу — два разных состояния. Блок 11.

ОТКУДА ВЗЯЛСЯ ЭТОТ БЛОК. Из ошибки в соседнем замере. Скрипт про отказ мастера
поднимал кластер, дожидался `cluster_state:ok` у всех узлов и ронял мастера.
Реплика не повышалась — ни через двадцать секунд, ни через минуту, никогда, — и
кластер оставался в `fail` навсегда. Выглядело как поломка замера, оказалось
поведением Valkey, про которое нигде не написано так, чтобы это бросалось в
глаза.

ЧТО ИМЕННО ПРОВЕРЯЕТСЯ. Между «слоты розданы» и «реплика способна заменить
мастера» есть окно в несколько секунд. Внутри него кластер отвечает
`cluster_state:ok` и обслуживает запросы, но отказ мастера он НЕ переживёт: у
реплики ещё не было ни одной успешной синхронизации, и она сама отказывается
повышаться.

ПОЧЕМУ ОТКАЗЫВАЕТСЯ — из исходника, а не из догадки. В `cluster_legacy.c`
возраст данных реплики считается так:

    if (server.repl_state == REPL_STATE_CONNECTED) {
        data_age = (server.unixtime - server.primary->last_interaction) * 1000;
    } else {
        data_age = (server.unixtime - server.repl_down_since) * 1000;
    }

У реплики, ни разу не подключавшейся, `repl_down_since` равен нулю — и возраст
данных получается отсчитанным от начала эпохи Unix. Дальше он сравнивается с
допуском `repl-ping-replica-period + cluster-node-timeout × validity-factor`, и
никакой допуск такого возраста не покрывает.

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

from __future__ import annotations

import os
import time

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

PAD = 34


def replica_of(ports: list[int], master_port: int) -> int | None:
    """Кто реплика указанного мастера — по CLUSTER NODES."""
    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 main() -> None:
    preamble("окно между «кластер собран» и «кластер переживёт отказ»")

    cluster = Cluster()
    try:
        ports = [free_port() for _ in range(6)]
        for p in ports:
            cluster.start_data(p, cluster=True)

        # Кластер создаётся БЕЗ ожидания готовности — в этом весь опыт.
        cluster.form_cluster_raw(ports, replicas=1)

        head(11, "СРАЗУ ПОСЛЕ СБОРКИ КЛАСТЕР РАБОТАЕТ, НО ОТКАЗ НЕ ПЕРЕЖИВЁТ")
        row("cluster-node-timeout", f"{CLUSTER_NODE_TIMEOUT_MS} мс", pad=PAD)
        row("cluster-replica-validity-factor",
            cli(ports[0], "config", "get", "cluster-replica-validity-factor").split()[-1], pad=PAD)
        row("repl-ping-replica-period",
            cli(ports[0], "config", "get", "repl-ping-replica-period").split()[-1], pad=PAD)
        print()

        # Первая часть: два признака готовности расходятся во времени.
        print("  как расходятся два признака готовности:")
        t0 = time.time()
        slots_at = None
        links_at = None
        while time.time() - t0 < 60 and (slots_at is None or links_at is None):
            if slots_at is None and all(cluster_state(p) == "ok" for p in ports):
                slots_at = time.time() - t0
                print(f"    t={slots_at:5.2f}  cluster_state:ok у всех узлов "
                      f"(слоты покрыты, запросы обслуживаются)")
            reps = [p for p in ports if role_of(p) == "slave"]
            if links_at is None and reps and all(link_status(p) == "up" for p in reps):
                links_at = time.time() - t0
                print(f"    t={links_at:5.2f}  master_link_status:up у всех реплик "
                      f"(есть кому повышаться)")
            time.sleep(0.1)
        print()
        if slots_at is not None and links_at is not None:
            row("окно между ними", f"{links_at - slots_at:.2f} с", pad=PAD)

        note(
            """
            Внутри этого окна кластер выглядит полностью рабочим: слоты
            покрыты, CLUSTER INFO отвечает ok, ключи пишутся и читаются. Но
            у реплик ещё нет ни одной завершённой синхронизации, и
            master_link_status у них down.
            """
        )

        # Вторая часть: то же окно, но с отказом внутри него — на свежем
        # кластере, чтобы окно было настоящим, а не воспроизведённым.
        head(12, "ЧТО БУДЕТ, ЕСЛИ МАСТЕР УПАДЁТ ВНУТРИ ЭТОГО ОКНА")
        for p in ports:
            cluster.kill(p)
        ports2 = [free_port() for _ in range(6)]
        for p in ports2:
            cluster.start_data(p, cluster=True)
        cluster.form_cluster_raw(ports2, replicas=1)

        # Ждём ровно первого признака и роняем мастера, не дожидаясь второго.
        deadline = time.time() + 60
        while time.time() < deadline and not all(cluster_state(p) == "ok" for p in ports2):
            time.sleep(0.05)
        masters = [p for p in ports2 if role_of(p) == "master"]
        victim = masters[0]
        rep = replica_of(ports2, 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)
        row("master_link_down_since_seconds", link_down_since(rep), pad=PAD)
        print()

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

        print()
        row("реплика повысилась", "да" if promoted else "НЕТ, за 40 с", pad=PAD)
        print()
        print("  что реплика записала в журнал:")
        log = os.path.join(cluster.dir, f"node-{rep}.log")
        if os.path.exists(log):
            seen = set()
            for line in open(log, encoding="utf-8").read().splitlines():
                if "unable to failover" in line or "Disconnected from primary" in line:
                    text = line.split(" * ")[-1].split(" # ")[-1]
                    if text not in seen:
                        # Строка длиннее блока: переносится по словам, чтобы
                        # в статью попала целиком, а не обрубком.
                        words, line_out = text.split(), "   "
                        for w in words:
                            if len(line_out) + 1 + len(w) > 68:
                                print(line_out)
                                line_out = "     " + w
                            else:
                                line_out += " " + w
                        print(line_out)
                        seen.add(text)

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

            Практический вывод не про Valkey, а про эксплуатацию: «кластер
            поднялся» и «кластер держит отказ» — разные проверки. Первая
            отвечает на вопрос, обслуживаются ли запросы; вторая требует
            смотреть на master_link_status каждой реплики. Автоматика,
            которая раскатывает кластер и сразу гасит старый, по первой
            проверке пройдёт, а по второй — нет.
            """
        )
    finally:
        cluster.stop_all()


if __name__ == "__main__":
    main()