"""Что видит клиент, пока падает мастер — блоки 8-10.

ГЛАВНЫЙ ВОПРОС. «Отказоустойчивость» в обоих режимах есть, но переживает
клиент эти секунды по-разному, и разница не в длительности, а в ТЕКСТЕ, который
он получает. В кластере ему отвечает живой узел — и отвечает так, что клиент
понимает, что делать. Под sentinel ему не отвечает никто: старый адрес просто
молчит, а новый клиент узнает, только если сходит спросить.

ПОЧЕМУ ЗДЕСЬ ОПРАШИВАЕТСЯ СЕРВЕР, А НЕ КЛИЕНТСКАЯ БИБЛИОТЕКА. Библиотека
прячет ровно то, что здесь интересно: она сама повторяет, сама ходит к
sentinel, сама ждёт. Опрос идёт «глупым» клиентом раз в 100 мс, и печатается
дословно то, что пришло по проводу.

ЧТО ЗДЕСЬ ИЗМЕРЯЕТСЯ. Секунды до восстановления — измеряются, но они целиком
про настройки этого замера (узел считается упавшим через секунду молчания
вместо тридцати по умолчанию). Переносится не число, а ПОРЯДОК событий и
текст ответов.

ЗАПУСК: python3 bench/valkey-cluster/failover.py
Занимает пару минут: переключения надо дождаться дважды.
"""

from __future__ import annotations

import time

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

SENTINEL_NAME = "cache"
KEY = "foo"
POLL_S = 0.1
PAD = 30


def owner_of(port: int, key: str) -> str | None:
    """Кому принадлежит слот ключа, по мнению узла port."""
    answer = cli(port, "set", key, "1")
    if answer.startswith("MOVED"):
        return answer.split()[-1]
    if answer == "OK":
        return f"127.0.0.1:{port}"
    return None


def watch(probe_port: int, key: str, seconds: float, label: str) -> list[tuple[float, str]]:
    """Опрашивать один адрес раз в POLL_S и записывать СМЕНЫ ответа.

    Печатать каждый опрос бессмысленно — их сотни. Содержательны моменты,
    когда ответ изменился: именно их и видит клиент как «что-то произошло».
    """
    out: list[tuple[float, str]] = []
    t0 = time.time()
    last = None
    while time.time() - t0 < seconds:
        answer = cli(probe_port, "set", key, "1", timeout=2.0) or "(пусто)"
        answer = answer.splitlines()[0][:46]
        if answer != last:
            out.append((time.time() - t0, answer))
            last = answer
        time.sleep(POLL_S)
    _ = label
    return out


def block8(cluster: Cluster, ports: list[int]) -> None:
    head(8, "ОТКАЗ МАСТЕРА В КЛАСТЕРЕ ГЛАЗАМИ КЛИЕНТА")

    owner = owner_of(ports[0], KEY)
    owner_port = int(owner.split(":")[1])
    # Опрашивается ЖИВОЙ узел, не тот, что упадёт: вопрос статьи не «что
    # отвечает мёртвый адрес» (ничего), а «что узнаёт клиент, добравшийся до
    # живого узла».
    probe = next(p for p in ports if p != owner_port and role_of(p) == "master")

    row("слот ключа принадлежит", owner, pad=PAD)
    row("опрашивается живой узел", f"127.0.0.1:{probe}", pad=PAD)
    row("cluster-node-timeout", f"{CLUSTER_NODE_TIMEOUT_MS} мс", pad=PAD)
    print()

    cluster.kill(owner_port)
    print(f"  t=0.00  узел {owner_port} убит")
    for t, answer in watch(probe, KEY, 20.0, "cluster"):
        print(f"  t={t:5.2f}  {answer}")

    new_owner = owner_of(probe, KEY)
    print()
    row("слот перешёл к", new_owner or "никому", pad=PAD)

    # Вторая половина блока: поднять убитый узел обратно и посмотреть, кем он
    # себя считает. Это прямой ответ на то, что в режиме Sentinel оказалось
    # самым опасным местом (блок 10), и проверяется он здесь, а не берётся на
    # веру.
    print()
    cluster.start_data(owner_port, cluster=True)
    time.sleep(2.0)
    row("вернувшийся узел: его роль", role_of(owner_port), pad=PAD)
    row("вернувшийся узел: SET foo", cli(owner_port, "set", KEY, "1"), pad=PAD)

    note(
        """
        Клиент всё это время получал от живого узла ОСМЫСЛЕННЫЕ ответы: пока
        переключение не произошло — старый адрес слота, после него — новый.
        Гадать не пришлось ни секунды: узел, до которого клиент дозвонился,
        знает состояние всего кластера, потому что узлы непрерывно
        обмениваются им по шине. (Если заглянуть в промежуток поточнее, там
        бывает и CLUSTERDOWN — тоже осмысленный ответ: «слот сейчас не
        покрыт никем».)

        Заметьте, чего здесь НЕ произошло: никто не спрашивал внешнюю
        службу. Реплику повысили сами узлы, большинством голосов мастеров,
        и они же разнесли новость. Кластер сам себе и наблюдатель, и арбитр.

        Две последние строки — про вернувшийся узел, и их стоит запомнить до
        блока 10. Поднявшись, он не стал ни мастером своих слотов, ни местом,
        куда можно записать: из шины он узнал, что слоты уже у другого, и
        сразу отдаёт клиенту адрес нового владельца.
        """
    )


def block9(cluster: Cluster, master: int, replicas: list[int], sentinels: list[int]) -> None:
    head(9, "ОТКАЗ МАСТЕРА ПОД SENTINEL ГЛАЗАМИ КЛИЕНТА")

    row("мастер по мнению sentinel", f"127.0.0.1:{master}", pad=PAD)
    row("down-after-milliseconds", f"{DOWN_AFTER_MS} мс", pad=PAD)
    print()

    t0 = time.time()
    cluster.kill(master)
    print(f"  t=0.00  узел {master} убит")

    # Две стороны опрашиваются одновременно по очереди: что отвечает старый
    # адрес и что говорит sentinel.
    seen_addr = None
    last_old = None
    deadline = time.time() + 25
    while time.time() < deadline:
        old = cli(master, "set", KEY, "1", timeout=1.0) or "(молчит)"
        old = old.splitlines()[0][:46]
        if old != last_old:
            print(f"  t={time.time() - t0:5.2f}  старый адрес: {old}")
            last_old = old
        addr = master_addr(sentinels[0], SENTINEL_NAME)
        if addr and addr[1] != master:
            if addr != seen_addr:
                print(f"  t={time.time() - t0:5.2f}  sentinel называет мастером {addr[0]}:{addr[1]}")
                seen_addr = addr
            break
        time.sleep(POLL_S)

    if seen_addr:
        new_port = seen_addr[1]
        time.sleep(1.0)
        row("новый мастер отвечает на запись", cli(new_port, "set", KEY, "1"), pad=PAD)
        other = [r for r in replicas if r != new_port]
        if other:
            row(f"вторая реплика {other[0]}", cli(other[0], "set", KEY, "1"), pad=PAD)

    print()
    print("  что sentinel записал у себя в журнале:")
    for line in cluster.sentinel_log(sentinels[0]):
        for mark in ("+sdown master", "+odown master", "+try-failover", "+vote-for-leader",
                     "+elected-leader", "+promoted-slave", "+failover-state", "+switch-master"):
            if mark in line:
                print("    " + line.split(" # ")[-1].split(" * ")[-1][:64])
                break

    note(
        """
        Разница с предыдущим блоком не в секундах, а в том, ЧТО получил
        клиент. Старый адрес не отвечал вовсе — там никого нет. Ни один
        узел данных не сказал клиенту нового адреса и не мог: узлы под
        sentinel про свою пару не знают, а служба, которая знает, клиенту
        сама не звонит. Клиент обязан спросить сам.

        В журнале видна и вся процедура: сначала один sentinel решает, что
        мастер молчит (+sdown), потом кворум подтверждает (+odown), потом
        выбирается ведущий (+vote-for-leader, +elected-leader) и только он
        повышает реплику. Два шага — «сломалось» и «кто чинит» — здесь
        разнесены намеренно, и в следующем блоке видно, зачем.
        """
    )


def block10(cluster: Cluster, old_master: int, sentinels: list[int]) -> None:
    head(10, "СТАРЫЙ МАСТЕР ВЕРНУЛСЯ: САМОЕ ОПАСНОЕ МЕСТО РЕЖИМА SENTINEL")

    current = master_addr(sentinels[0], SENTINEL_NAME)
    row("мастер сейчас", f"{current[0]}:{current[1]}" if current else "неизвестен", pad=PAD)
    print()

    t0 = time.time()
    cluster.start_data(old_master)
    print(f"  t=0.00  узел {old_master} поднят обратно")

    # Ключевой вопрос: как долго вернувшийся узел считает себя мастером и
    # принимает записи, хотя мастер уже не он.
    demoted_at = None
    last = None
    deadline = time.time() + 30
    while time.time() < deadline:
        role = role_of(old_master)
        write = cli(old_master, "set", "who", "old", timeout=1.0)
        state = f"роль {role}, запись -> {write}"
        if state != last:
            print(f"  t={time.time() - t0:5.2f}  {state}")
            last = state
        if role == "slave":
            demoted_at = time.time() - t0
            break
        time.sleep(POLL_S)

    print()
    if demoted_at is None:
        row("за 30 с в реплику НЕ переведён", "", pad=PAD)
    else:
        row("переведён в реплику через", f"{demoted_at:.2f} с", pad=PAD)
    # Смотреть сразу после перевода нельзя: синхронизация с настоящим мастером
    # ещё идёт, и ключ в этот момент на месте. Ответ на вопрос «что стало с
    # записью» даёт только состояние ПОСЛЕ синхронизации.
    row("сразу после перевода: GET who", cli(old_master, "get", "who") or "(её нет)", pad=PAD)
    deadline = time.time() + 20
    while time.time() < deadline and link_status(old_master) != "up":
        time.sleep(0.2)
    time.sleep(1.0)
    row("после синхронизации: GET who", cli(old_master, "get", "who") or "(её нет)", pad=PAD)

    note(
        """
        Вот окно, ради которого блок и написан. Вернувшийся узел поднимается
        МАСТЕРОМ — он таким себя помнит — и принимает записи. Sentinel
        переводит его в реплику не мгновенно, а обнаружив.

        Две последние строки надо читать вместе, и именно в таком порядке.
        Сразу после перевода запись ещё на месте: узел уже реплика, но
        синхронизация только началась. А после неё ключа нет — данные
        заменены копией настоящего мастера. То есть клиент получил на свою
        запись OK, и запись исчезла; никакой ошибки он при этом не видел.

        Опасность не в самом окне, а в том, кто в него попадает: клиент,
        который держит адрес данных напрямую и не спрашивает sentinel. Для
        него старый адрес снова живой, отвечает OK — и всё идёт в никуда.
        В кластере такого окна нет: вернувшийся узел узнаёт из шины, что
        его слоты уже у другого, ещё до того, как примет первую команду.
        """
    )


def main() -> None:
    preamble("отказ мастера в обоих режимах: что приходит клиенту по проводу")

    cluster = Cluster()
    try:
        cports = [free_port() for _ in range(6)]
        for p in cports:
            cluster.start_data(p, cluster=True)
        cluster.form_cluster(cports, replicas=1)
        block8(cluster, cports)

        m = free_port()
        reps = [free_port(), free_port()]
        sents = [free_port() for _ in range(3)]
        cluster.start_data(m)
        for r in reps:
            cluster.start_data(r, replicaof=m)
        for s in sents:
            cluster.start_sentinel(s, m, quorum=2, name=SENTINEL_NAME)
        time.sleep(3)
        cli(m, "set", KEY, "1")

        block9(cluster, m, reps, sents)
        block10(cluster, m, sents)
    finally:
        cluster.stop_all()


if __name__ == "__main__":
    main()