"""Нумерованные базы в кластере: версия Valkey меняет ответ — блоки 14-16.

ОТКУДА ВЗЯЛСЯ ЭТОТ СКРИПТ. Из ошибки в первой версии статьи. Там стояло
«в кластере база ровно одна», и подтверждалось это ответом сервера:

    SELECT 1  →  ERR DB index is out of range

Ответ настоящий, вывод неверный. Это правило Redis и Valkey до 9.0; в Valkey
9.0 появились нумерованные базы в кластере, а число их задаёт отдельная
настройка `cluster-databases` со значением по умолчанию 1. То есть замер
показывал не отсутствие возможности, а НАСТРОЙКУ ПО УМОЛЧАНИЮ — и отличить
одно от другого можно было только вторым опытом, которого не было.

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

ПОЧЕМУ `cluster-databases` НЕЛЬЗЯ ПОМЕНЯТЬ НА ЛЕТУ. В исходнике настройка
объявлена IMMUTABLE_CONFIG (`src/config.c`), поэтому оба кластера поднимаются
отдельно, а не перенастраиваются.

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

from __future__ import annotations

import time

from common import Cluster, cli, free_port, head, note, preamble, row

PAD = 40


def owner_of(ports: list[int], key: str) -> int:
    """Узел, которому принадлежит слот ключа."""
    answer = cli(ports[0], "set", key, "probe")
    if answer.startswith("MOVED"):
        return int(answer.split()[-1].split(":")[1])
    return ports[0]


def spin(cluster: Cluster, databases: int) -> list[int]:
    """Поднять кластер из трёх мастеров с заданным числом баз."""
    ports = [free_port() for _ in range(3)]
    for p in ports:
        cluster.start_data(p, cluster=True, extra=["--cluster-databases", str(databases)])
    cluster.form_cluster_raw(ports, replicas=0)
    deadline = time.time() + 60
    from common import cluster_state
    while time.time() < deadline:
        if all(cluster_state(p) == "ok" for p in ports):
            break
        time.sleep(0.2)
    return ports


def block14(default_ports: list[int], many_ports: list[int]) -> None:
    head(14, "СКОЛЬКО БАЗ В КЛАСТЕРЕ, РЕШАЕТ НАСТРОЙКА, А НЕ РЕЖИМ")

    d_owner = owner_of(default_ports, "foo")
    m_owner = owner_of(many_ports, "foo")

    row("кластер А: cluster-databases",
        cli(default_ports[0], "config", "get", "cluster-databases").split()[-1], pad=PAD)
    row("кластер Б: cluster-databases",
        cli(many_ports[0], "config", "get", "cluster-databases").split()[-1], pad=PAD)
    print()
    for label, port in (("кластер А (по умолчанию)", d_owner), ("кластер Б (16 баз)", m_owner)):
        row(f"{label}: SELECT 0", cli(port, "select", "0"), pad=PAD)
        row(f"{label}: SELECT 1", cli(port, "select", "1"), pad=PAD)
        row(f"{label}: SELECT 15", cli(port, "select", "15"), pad=PAD)
        row(f"{label}: SELECT 16", cli(port, "select", "16"), pad=PAD)
        print()

    note(
        """
        Два кластера отличаются одной строкой конфига, и ответы на одну и ту
        же команду у них разные. Значит, `ERR DB index is out of range` в
        кластере А — не «в кластере баз не бывает», а «в ЭТОМ кластере база
        настроена одна».

        Разница между этими двумя прочтениями и есть содержание блока. Первое
        — правило Redis и Valkey до 9.0, и оно до сих пор кочует по статьям.
        Второе — то, что на самом деле происходит на Valkey 9.

        Обратите внимание и на то, что НЕ изменилось: значение по умолчанию
        осталось единицей. Кластер, поднятый обычным способом, ведёт себя
        по-старому, и именно поэтому старое правило так долго выглядит
        верным.
        """
    )


def block15(many_ports: list[int]) -> None:
    head(15, "НОМЕР БАЗЫ НЕ УЧАСТВУЕТ В ВЫЧИСЛЕНИИ СЛОТА")

    owner = owner_of(many_ports, "foo")

    row("узел-владелец слота ключа foo", f"127.0.0.1:{owner}", pad=PAD)
    print()
    row("в базе 0: SET foo db0", cli(owner, "-n", "0", "set", "foo", "db0"), pad=PAD)
    row("в базе 1: SET foo db1", cli(owner, "-n", "1", "set", "foo", "db1"), pad=PAD)
    print()
    row("в базе 0: GET foo", cli(owner, "-n", "0", "get", "foo"), pad=PAD)
    row("в базе 1: GET foo", cli(owner, "-n", "1", "get", "foo"), pad=PAD)
    print()
    row("в базе 0: CLUSTER KEYSLOT foo", cli(owner, "-n", "0", "cluster", "keyslot", "foo"), pad=PAD)
    row("в базе 1: CLUSTER KEYSLOT foo", cli(owner, "-n", "1", "cluster", "keyslot", "foo"), pad=PAD)

    note(
        """
        Один ключ, два значения, один слот. Это и есть модель данных Valkey 9:
        адресация по-прежнему двухшаговая — ключ в слот, слот в узел, — а база
        добавляется ТРЕТЬИМ измерением уже внутри узла и на выбор узла не
        влияет вовсе.

        Практическое следствие приятное: разнести окружения по базам в
        кластере можно, и раскладка данных по узлам от этого не изменится ни
        на ключ. Практическое следствие неприятное — в следующем блоке.
        """
    )


def block16(many_ports: list[int]) -> None:
    head(16, "ЧТО НЕ ИЗМЕНИЛОСЬ: ЗАПРЕТЫ И ГРАНИЦА ВИДИМОСТИ")

    owner = owner_of(many_ports, "foo")
    other = [p for p in many_ports if p != owner]

    row("SWAPDB 0 1", cli(owner, "swapdb", "0", "1"), pad=PAD)
    print()
    # Ключи раскладываются по РАЗНЫМ слотам нарочно: вопрос блока — что видит
    # команда, выполняемая на одном узле, когда данные лежат на трёх.
    for key, value in (("foo", "1"), ("bar", "2"), ("user:1", "3")):
        for port in many_ports:
            if not cli(port, "-n", "1", "set", key, value).startswith("MOVED"):
                break
    print("  в базе 1 записаны три ключа, разошедшиеся по трём узлам:")
    for key in ("foo", "bar", "user:1"):
        row(f"    CLUSTER KEYSLOT {key}", cli(owner, "cluster", "keyslot", key), pad=PAD)
    print()
    print("  что отвечает DBSIZE в базе 1 на каждом узле:")
    for port in many_ports:
        row(f"    узел {port}", cli(port, "-n", "1", "dbsize"), pad=PAD)
    print()
    row("SCAN в базе 1 на узле-владельце", " ".join(cli(owner, "-n", "1", "scan", "0").split()), pad=PAD)
    if other:
        row(f"SCAN в базе 1 на узле {other[0]}",
            " ".join(cli(other[0], "-n", "1", "scan", "0").split()), pad=PAD)

    note(
        """
        Три вещи разом, и все три — против обратного мифа «раз базы появились,
        они работают как в одиночном узле».

        Первое: SWAPDB как был запрещён, так и остался, и текст отказа прямо
        называет причину — режим. Обменять базы на одном шарде значило бы
        рассогласовать его с остальными.

        Второе и главное: DBSIZE и SCAN считают не базу целиком, а её часть НА
        ТОМ УЗЛЕ, куда пришла команда. Три ключа записаны в одну базу и
        разошлись по трём узлам — и ни один узел не видит всех трёх. То же
        касается FLUSHDB.

        Отсюда правило, которое стоит унести: нумерованная база в кластере —
        это пространство имён, а не отдельное хранилище. Она не даёт ни
        изоляции ресурсов, ни общего на кластер представления о своём
        содержимом.
        """
    )


def main() -> None:
    preamble("два кластера, отличающиеся одной строкой конфига: cluster-databases 1 и 16")

    cluster = Cluster()
    try:
        default_ports = spin(cluster, 1)
        many_ports = spin(cluster, 16)
        block14(default_ports, many_ports)
        block15(many_ports)
        block16(many_ports)
    finally:
        cluster.stop_all()


if __name__ == "__main__":
    main()