test_versiondb.py

import shutil
import tempfile
from pathlib import Path

import pytest
import tomlkit
from pystarport.ports import rpc_port

from .utils import (
    BASECRO_DENOM,
    cluster_fixture,
    wait_for_block,
    wait_for_new_blocks,
    wait_for_port,
)

pytestmark = pytest.mark.normal


def cluster(worker_index, tmp_path_factory):
    "override cluster fixture for this test module"
    yield from cluster_fixture(
        Path(__file__).parent / "configs/default.jsonnet",
        worker_index,
        tmp_path_factory.mktemp("data"),
    )


def test_versiondb_migration(cluster):
    """
    test versiondb migration commands.
    node0 has memiavl and versiondb enabled while node1 don't,
    - stop all the nodes
    - dump change set from node1's application.db
    - verify change set and save snapshot
    - restore pruned application.db from the snapshot
    - replace node1's application.db with the restored one
    - rebuild versiondb for node0
    - start the nodes, now check:
      - the network can grow
      - node0 do support historical queries
      - node1 don't support historical queries
    """
    community_addr = cluster.address("community")
    reserve_addr = cluster.address("reserve")

    block0 = cluster.block_height()
    cm_balance0 = cluster.balance(community_addr)
    rs_balance0 = cluster.balance(reserve_addr)

    cro_1 = 100000000
    coins_1 = f"{cro_1}{BASECRO_DENOM}"

    rsp = cluster.transfer(community_addr, reserve_addr, coins_1)
    assert rsp["code"] == 0, rsp

    block1 = cluster.block_height()
    cm_balance1 = cluster.balance(community_addr)
    rs_balance1 = cluster.balance(reserve_addr)

    assert cm_balance1 == cm_balance0 - cro_1
    assert rs_balance1 == rs_balance0 + cro_1

    # wait for a few blocks
    wait_for_new_blocks(cluster, 2)

    # stop the network first
    print("stop all nodes")
    cluster.supervisor.stopAllProcesses()

    # check the state of all nodes should be stopped
    for info in cluster.supervisor.getAllProcessInfo():
        assert info["statename"] == "STOPPED"

    node0 = cluster.cosmos_cli(i=0)
    node1 = cluster.cosmos_cli(i=1)
    # dump change set from node1's application.db
    changeset_dir = tempfile.mkdtemp(dir=cluster.data_root)
    print("dump to:", changeset_dir)
    # only restore to an intermediate version to test version mismatch behavior
    print(node1.changeset_dump(changeset_dir, end_version=block1 + 1))

    # verify and save to snapshot
    snapshot_dir = tempfile.mkdtemp(dir=cluster.data_root)
    print("verify and save to snapshot:", snapshot_dir)
    _, commit_info = node0.changeset_verify(changeset_dir, save_snapshot=snapshot_dir)
    latest_version = commit_info["version"]

    # replace existing `application.db`
    app_db1 = node1.data_dir / "data/application.db"
    print("replace node db:", app_db1)
    shutil.rmtree(app_db1)
    print(node1.changeset_restore_app_db(snapshot_dir, app_db1))

    # rebuild versiondb to verify the versiondb can be rebuilt from the changeset
    print("rebuilding versiondb for node0 from changeset")
    sst_dir = tempfile.mkdtemp(dir=cluster.data_root)
    print(node0.changeset_build_versiondb_sst(changeset_dir, sst_dir))

    # ingest-versiondb-sst expects an empty database
    versiondb_path = node0.data_dir / "data/versiondb"
    if versiondb_path.exists():
        shutil.rmtree(versiondb_path)
    print(
        node0.changeset_ingest_versiondb_sst(
            versiondb_path, sst_dir, maximum_version=latest_version
        )
    )

    # force node1's app-db-backend to be rocksdb
    app_toml_path = node1.data_dir / "config/app.toml"
    patch_app_db_backend(app_toml_path, "rocksdb")

    start_all_nodes(cluster)
    # wait for consensus to be stable after restart
    wait_for_new_blocks(cluster, 2)
    # ensure both nodes have replayed up to block1 before historical queries
    wait_for_block(node0, block1)
    wait_for_block(node1, block1)

    # node0 supports historical queries with versiondb
    assert node0.balance(community_addr, height=block0) == cm_balance0
    assert node0.balance(community_addr, height=block1) == cm_balance1
    assert cluster.balance(community_addr) == cm_balance1

    # check query still works, node1 don't enable versiondb,
    # so we are testing iavl query here.
    assert node1.balance(community_addr) == cm_balance1

    # Verify node1 cannot support historical queries
    with pytest.raises(Exception):
        # This should fail because node1 doesn't have versiondb enabled
        node1.balance(community_addr, height=block0)

    # check the chain is still growing
    rsp = cluster.transfer(community_addr, reserve_addr, coins_1)
    # avoid the tx fails
    assert rsp["code"] == 0, rsp

    cm_balance2 = cluster.balance(community_addr)
    rs_balance2 = cluster.balance(reserve_addr)

    assert cm_balance2 == cm_balance1 - cro_1
    assert rs_balance2 == rs_balance1 + cro_1

    # now we restore node1's db again and test it with memiavl enabled
    cluster.supervisor.stopAllProcesses()

    # restore application.db from the snapshot
    print("replace node db:", app_db1)
    shutil.rmtree(app_db1)
    print(node1.changeset_restore_app_db(snapshot_dir, app_db1))

    patch_app_memiavl_enabled(app_toml_path, "true")

    start_all_nodes(cluster)
    # wait for consensus to be stable after restart
    wait_for_new_blocks(cluster, 2)
    # ensure node1 has replayed up to block1 before historical queries
    wait_for_block(node1, block1)
    # should be able to query node1's balance with memiavl
    assert node1.balance(community_addr, height=block0) == cm_balance0
    assert node1.balance(community_addr, height=block1) == cm_balance1


def start_all_nodes(cluster, validator_count=2):
    print("start all nodes")
    for i in range(validator_count):
        cluster.supervisor.startProcess(f"{cluster.chain_id}-node{i}")

    for i in range(validator_count):
        wait_for_port(rpc_port(cluster.base_port(i)))


def patch_app_db_backend(path, backend):
    cfg = tomlkit.parse(path.read_text())
    cfg["app-db-backend"] = backend
    path.write_text(tomlkit.dumps(cfg))


def patch_app_memiavl_enabled(path, enabled):
    cfg = tomlkit.parse(path.read_text())
    cfg["memiavl"]["enable"] = enabled
    path.write_text(tomlkit.dumps(cfg))