diff --git a/README.md b/README.md index acd6a56..ee70a82 100644 --- a/README.md +++ b/README.md @@ -67,4 +67,49 @@ class TestExamplePerClassSetup(ExampleTestCaseBase): client.execute_command("SET K V") ``` -For more examples, refer to the `tests` directory of this package. +**Testing Cluster Mode Enabled (CME)** + +To test against a Valkey cluster, inherit `ClusterTestCase` instead of `ValkeyTestCase`. Call `setup_cluster(num_shards, num_replicas_per_shard)` to bootstrap a cluster: it starts all nodes, assigns slots, attaches replicas, waits until the cluster state is `ok`, and returns a client that follows `MOVED`/`ASK` redirections. The cluster is torn down automatically after each test. + +``` +from valkey_test_case import ClusterTestCase, ClusterInfo + +class TestExampleCluster(ClusterTestCase): + def test_cluster_read_write(self): + self.server_path = "/path_to_your_valkey_server_binary" + + # 3 primaries, each with 1 replica (6 nodes total) + client = self.setup_cluster(num_shards=3, num_replicas_per_shard=1) + + client.set("key", "value") + assert client.get("key") == b"value" + + # self.nodes holds every ClusterNodeHandle for direct inspection + for node in self.nodes: + assert ClusterInfo(node.client.cluster("INFO")).is_cluster_ok() +``` + +To apply startup arguments (modules, configs) to every node, set `self.args` inside the test before calling `setup_cluster`. + +To test slot migration, use `migrate_slot(source, target, slot)` to move a slot and its keys from one node to another, then `wait_for_slot_owner(slot, target)` to wait until every node agrees on the new owner. `get_slot_owner(slot)` returns the node that currently owns a slot. + +``` +from valkey_test_case import ClusterTestCase +from valkey.cluster import key_slot + +class TestExampleMigration(ClusterTestCase): + def test_migrate(self): + self.server_path = "/path_to_your_valkey_server_binary" + self.setup_cluster(num_shards=2, num_replicas_per_shard=0) + + slot = key_slot(b"key") + source = self.get_slot_owner(slot) + target = next(n for n in self.nodes if n.nodeid != source.nodeid) + source.client.set("key", "value") + + self.migrate_slot(source, target, slot) + self.wait_for_slot_owner(slot, target) + assert target.client.get("key") == b"value" +``` + +Pass `dbs=(0, 1, ...)` to `migrate_slot` to move keys across multiple databases; migrating any database other than 0 requires the cluster to be started with `cluster-databases > 1`. diff --git a/src/valkey_test_case.py b/src/valkey_test_case.py index 90a3470..cfdb12c 100644 --- a/src/valkey_test_case.py +++ b/src/valkey_test_case.py @@ -8,7 +8,7 @@ from functools import wraps from valkey import * from util.waiters import * - +from valkey.cluster import VALKEY_CLUSTER_HASH_SLOTS, ValkeyCluster from enum import Enum MAX_PING_TRIES = 60 @@ -728,3 +728,364 @@ def waitForReplicaOffsetToSyncUp(self, primary, replica): pinfo.get_primary_repl_offset(), timeout=TEST_MAX_WAIT_TIME_SECONDS, ) + + +class ClusterInfo: + """Contains information about a point in time of a Valkey cluster.""" + + def __init__(self, info): + self.info = info + + def is_cluster_ok(self): + """Return True if the cluster state is OK.""" + return self.info["cluster_state"] == "ok" + + def cluster_known_nodes(self): + """Return the number of nodes known to this node.""" + return int(self.info["cluster_known_nodes"]) + + def cluster_slots_assigned(self): + """Return the number of hash slots currently assigned.""" + return int(self.info["cluster_slots_assigned"]) + + +class ClusterNodeHandle(ValkeyServerHandle): + """Handle to a valkey server process running in cluster mode enabled (CME).""" + + def __init__( + self, + bind_ip, + port, + port_tracker, + testdir, + server_path="valkey-server", + ): + super(ClusterNodeHandle, self).__init__( + bind_ip, port, port_tracker, server_path=server_path, cwd=testdir + ) + # Start the node in cluster mode. The cluster-config-file (nodes.conf) + # is per-node and cleaned up by the base class teardown. + self.args["cluster-enabled"] = "yes" + self.args["cluster-config-file"] = "nodes_{}_{}.conf".format(bind_ip, port) + self.args["cluster-node-timeout"] = "2000" + self.masterid = None + self.nodeid = None + + def _set_node_id(self): + # CLUSTER NODES should return only one node - myself after startup + # Read the node id + nodes = self.client.cluster("NODES") + for key in nodes: + if re.match("myself", nodes[key]["flags"]): + self.nodeid = nodes[key]["node_id"] + logging.info( + "Cluster node {} has node id {}".format(self.port, self.nodeid) + ) + return + + def connect(self): + client = super(ClusterNodeHandle, self).connect() + self._set_node_id() + return client + + def meet(self, ip, port): + return self.client.execute_command("CLUSTER", "MEET", ip, port) + + def replicate(self, node_id): + self.masterid = node_id + return self.client.execute_command("CLUSTER", "REPLICATE", node_id) + + def assign_slots(self, *args): + """ + Assign multiple ranges of slots to this node. + Accepts multiple pairs that are interpretted as [low,high) + assign_slots(0,10) -> [0..9] + assign_slots(0,10,15,20) -> [0..9] and [15..19] + """ + assert len(args) % 2 == 0 + command = ["CLUSTER", "ADDSLOTSRANGE"] + for t in range(0, len(args), 2): + command.extend([args[t], args[t + 1] - 1]) + return self.client.execute_command(*command) + + # Building blocks for moving a slot from one node to another. + # ClusterTestCase.migrate_slot() calls these in the right order. + + def count_keys_in_slot(self, slot): + """How many keys this node currently holds in the given slot.""" + return int(self.client.execute_command("CLUSTER", "COUNTKEYSINSLOT", slot)) + + def start_importing_slot(self, slot, source_id): + """Tell this node to start accepting a slot coming from source_id.""" + return self.client.execute_command( + "CLUSTER", "SETSLOT", slot, "IMPORTING", source_id + ) + + def start_migrating_slot(self, slot, target_id): + """Tell this node it is handing a slot over to target_id.""" + return self.client.execute_command( + "CLUSTER", "SETSLOT", slot, "MIGRATING", target_id + ) + + def assign_slot_owner(self, slot, owner_id): + """Record which node now owns the slot. Only valid on a primary.""" + return self.client.execute_command("CLUSTER", "SETSLOT", slot, "NODE", owner_id) + + def get_slot_owner_id(self, slot): + """Which node id THIS node believes owns the given slot (or None).""" + for slot_range in self.client.cluster("SLOTS"): + start, end = slot_range[0], slot_range[1] + if start <= slot <= end: + return slot_range[2][2].decode() + return None + + def is_primary(self): + """True if this node is currently a primary (master), not a replica.""" + role = self.client.execute_command("ROLE")[0] + return role in (b"master", "master") + + def wait_for_cluster_known_nodes(self, count): + """Wait until we are connected to exactly count nodes.""" + wait_for_equal( + lambda: ClusterInfo(self.client.cluster("INFO")).cluster_known_nodes(), + count, + timeout=TEST_MAX_WAIT_TIME_SECONDS, + ) + + def wait_for_cluster_known_node(self, nodeid): + def knows(): + nodesInfo = self.client.cluster("NODES") + for key in nodesInfo: + if nodeid == nodesInfo[key]["node_id"]: + return True + return False + + wait_for_true(knows, timeout=TEST_MAX_WAIT_TIME_SECONDS) + + def wait_for_cluster_ok(self): + wait_for_true( + lambda: ClusterInfo(self.client.cluster("INFO")).is_cluster_ok(), + timeout=TEST_MAX_WAIT_TIME_SECONDS, + ) + + +class ClusterTestCase(ValkeyTestCase): + """Base class for Cluster Mode Enabled (CME) tests.""" + + @pytest.fixture(autouse=True) + def cluster_setup(self): + # Per-test cluster state. Initialized here rather than as class-level + # attributes so each test starts with its own node list. + self.nodes = [] + self.cluster_client = None + yield + self.teardown() + + def create_node(self, bind_ip=None, port=None): + """Create a single cluster-mode node and register it for teardown.""" + if not bind_ip: + bind_ip = self.get_bind_ip() + if not port: + port = self.get_bind_port() + + node = ClusterNodeHandle( + bind_ip=bind_ip, + port=port, + port_tracker=self.port_tracker, + testdir=self.testdir, + server_path=self.server_path, + ) + node.args.update(self.args) + self.nodes.append(node) + # Registered in server_list so ValkeyTestCase.teardown reclaims it. + self.server_list.append(node) + return node + + def create_nodes(self, num_nodes): + for _ in range(num_nodes): + self.create_node() + return self.nodes + + def start_all_nodes(self, wait_for_ping=True, connect_client=True): + for node in self.nodes: + node.start(wait_for_ping=wait_for_ping, connect_client=connect_client) + + def create_cluster(self, num_nodes): + """Start `num_nodes` nodes and gossip them into a single cluster.""" + self.create_nodes(num_nodes) + self.start_all_nodes() + + # Introduce every other node to the first node; gossip propagates the + # full topology from there. + for i in range(1, num_nodes): + self.nodes[0].meet(self.nodes[i].bind_ip, self.nodes[i].port) + + # Wait until every node has discovered the whole cluster. + for node in self.nodes: + node.wait_for_cluster_known_nodes(num_nodes) + + def assign_slots_to_nodes(self, num_nodes): + """Equally distribute sequential slots to each node.""" + slot_slice = VALKEY_CLUSTER_HASH_SLOTS / num_nodes + for i in range(num_nodes): + slot_min = int(round(slot_slice * i)) + slot_max = int(round(slot_slice * (i + 1))) + self.nodes[i].assign_slots(slot_min, slot_max) + + def setup_replicas(self, num_shards, num_replicas_per_shard): + """Attach the remaining nodes as replicas, round-robin across shards.""" + total = num_shards * (1 + num_replicas_per_shard) + for i in range(num_shards, total): + shard_idx = i % num_shards + primary = self.nodes[shard_idx] + # Make sure the replica knows the primary before replicating. + self.nodes[i].wait_for_cluster_known_node(primary.nodeid) + self.nodes[i].replicate(primary.nodeid) + + # Wait for each shard's replicas to come online and sync up. + for i in range(num_shards): + wait_for_equal( + lambda primary=self.nodes[i]: primary.num_replicas_online(), + num_replicas_per_shard, + timeout=MAX_REPLICA_WAIT_TIME, + ) + for i in range(num_shards, total): + self.waitForReplicaToSyncUp(self.nodes[i]) + # Allow read-only queries to be served by the replica. + self.nodes[i].client.readonly() + + def setup_cluster(self, num_shards, num_replicas_per_shard): + """Create and fully bootstrap a cluster, returning a cluster client. + + When this returns the cluster is in the 'ok' state and ready to serve. + """ + total_nodes = num_shards * (1 + num_replicas_per_shard) + self.create_cluster(total_nodes) + + self.assign_slots_to_nodes(num_shards) + + # Bump each primary's config epoch so replicas don't overtake it via + # epoch collision resolution. + for i in range(num_shards): + self.nodes[i].client.execute_command("CLUSTER", "BUMPEPOCH") + + if num_replicas_per_shard > 0: + self.setup_replicas(num_shards, num_replicas_per_shard) + + # Wait for every node to agree the cluster is healthy. + for node in self.nodes: + node.wait_for_cluster_ok() + + self.cluster_client = self.get_cluster_client() + return self.cluster_client + + def get_cluster_client(self): + """Return a cluster-aware client that follows MOVED/ASK redirections.""" + primary = self.nodes[0] + # A cluster client discovers the topology by connecting to the host each + # node advertises in CLUSTER SLOTS, not the bind address. When a node + # binds the wildcard 0.0.0.0 it advertises a concrete, connectable host + # (typically loopback) instead, so read that advertised host back and + # seed the client with it rather than the bind address. + host = primary.bind_ip + for slot_range in primary.client.cluster("SLOTS"): + # slot_range = [start, end, [host, port, node_id, ...], ...] + owner_host, _, owner_id = ( + slot_range[2][0], + slot_range[2][1], + slot_range[2][2], + ) + if owner_id.decode() == primary.nodeid: + host = owner_host.decode() + break + return ValkeyCluster(host=host, port=primary.port) + + def get_slot_owner(self, slot): + """Return the node that currently owns `slot`, or None if unassigned.""" + for slot_range in self.nodes[0].client.cluster("SLOTS"): + start, end, owner_id = ( + slot_range[0], + slot_range[1], + slot_range[2][2].decode(), + ) + if start <= slot <= end: + return next((n for n in self.nodes if n.nodeid == owner_id), None) + return None + + def wait_for_slot_owner(self, slot, expected_owner): + """Wait until all nodes agree `expected_owner` owns `slot`. + + New ownership has to gossip across the cluster after a migration, so we + poll instead of sleeping for a fixed guess. + """ + wait_for_true( + lambda: all( + node.get_slot_owner_id(slot) == expected_owner.nodeid + for node in self.nodes + ), + timeout=TEST_MAX_WAIT_TIME_SECONDS, + ) + + def migrate_slot(self, source, target, slot, dbs=(0,), timeout_ms=5000): + """Move a slot and all its keys from source to target, then hand off + ownership. + + Runs the manual migration protocol: mark the slot IMPORTING on the + target and MIGRATING on the source, batch-move every key in the slot + with MIGRATE, then announce the new owner on every primary. Replicas + reject CLUSTER SETSLOT and learn the new owner from their primary, so + they are skipped. + + `dbs` lists which databases to move keys from. Migrating any DB other + than 0 requires the cluster to be started with `cluster-databases > 1`; + plain cluster mode only has DB 0. + """ + target.start_importing_slot(slot, source.nodeid) + source.start_migrating_slot(slot, target.nodeid) + + # Move keys on a dedicated connection so the caller's client keeps its + # own selected DB. GETKEYSINSLOT only sees the connection's current DB, + # so we select each DB in turn and drain the slot in batches. + conn = source.get_new_client() + try: + for db in dbs: + conn.execute_command("SELECT", db) + while True: + keys = conn.execute_command("CLUSTER", "GETKEYSINSLOT", slot, 100) + if not keys: + break + conn.execute_command( + "MIGRATE", + target.bind_ip, + target.port, + "", + db, + timeout_ms, + "KEYS", + *keys, + ) + finally: + conn.close() + + # Finalize ownership target-first: the target (and its replicas) must + # persist the new topology before the source gives up the slot. An + # out-of-order handoff where the source releases first could leave the + # slot ownerless if the target then fails. Remaining primaries are + # updated afterward; replicas learn from their primary / via gossip. + target.assign_slot_owner(slot, target.nodeid) + source.assign_slot_owner(slot, target.nodeid) + for node in self.nodes: + if node.is_primary() and node.nodeid not in ( + target.nodeid, + source.nodeid, + ): + node.assign_slot_owner(slot, target.nodeid) + + def teardown(self): + if self.cluster_client is not None: + try: + self.cluster_client.close() + except Exception: + pass + self.cluster_client = None + ValkeyTestCase.teardown(self) diff --git a/tests/test_cluster_mode.py b/tests/test_cluster_mode.py new file mode 100644 index 0000000..e51fe08 --- /dev/null +++ b/tests/test_cluster_mode.py @@ -0,0 +1,123 @@ +from conftest import resource_port_tracker +from valkey_test_case import ClusterTestCase, ClusterInfo +import pytest +import os + + +SERVER_PATH = os.path.join( + os.path.dirname(os.path.realpath(__file__)), + ".build", + "binaries", + os.environ["SERVER_VERSION"], + "valkey-server", +) + + +class TestClusterBasic(ClusterTestCase): + """Verify that ClusterTestCase can bootstrap a 3-primary cluster.""" + + server_path = SERVER_PATH + + @pytest.mark.order(1) + def test_setup_cluster_and_use(self): + client = self.setup_cluster(num_shards=3, num_replicas_per_shard=0) + + # Verify cluster_state is ok on all nodes + for node in self.nodes: + assert ClusterInfo( + node.client.cluster("INFO") + ).is_cluster_ok(), f"Node {node.port} not in ok state" + + # Verify all 16384 slots are assigned + info = ClusterInfo(self.nodes[0].client.cluster("INFO")) + assert info.cluster_slots_assigned() == 16384 + assert info.cluster_known_nodes() == 3 + + # Write and read keys — ValkeyCluster follows MOVED redirections + for i in range(100): + client.set(f"key:{i}", f"value:{i}") + + for i in range(100): + val = client.get(f"key:{i}") + assert val == b"value:" + str(i).encode() + + @pytest.mark.order(2) + @pytest.mark.parametrize("num_shards", [2, 3, 5]) + def test_slot_distribution(self, num_shards): + self.setup_cluster(num_shards=num_shards, num_replicas_per_shard=0) + + # Collect the set of slots each primary owns from CLUSTER SLOTS. + # Reply shape: [[start, end, [host, port, node_id, ...], ...], ...] + owned = {} + for slot_range in self.nodes[0].client.cluster("SLOTS"): + start, end = slot_range[0], slot_range[1] + owner_id = slot_range[2][2].decode() + owned.setdefault(owner_id, set()).update(range(start, end + 1)) + + # Every primary owns a share of the slots. + assert len(owned) == num_shards + + all_slots = set() + for slots in owned.values(): + # No slot is claimed by more than one primary. + assert all_slots.isdisjoint(slots) + all_slots |= slots + + # Every one of the 16384 slots is covered exactly once. + assert all_slots == set(range(16384)) + + +class TestClusterWithReplicas(ClusterTestCase): + """Verify multi-shard clusters with replicas.""" + + server_path = SERVER_PATH + + @pytest.mark.order(1) + def test_setup_cluster_with_replicas(self): + client = self.setup_cluster(num_shards=3, num_replicas_per_shard=1) + + # 6 nodes total + assert len(self.nodes) == 6 + + # Cluster is healthy + for node in self.nodes: + assert ClusterInfo( + node.client.cluster("INFO") + ).is_cluster_ok(), f"Node {node.port} not in ok state" + + # All slots assigned + info = ClusterInfo(self.nodes[0].client.cluster("INFO")) + assert info.cluster_slots_assigned() == 16384 + assert info.cluster_known_nodes() == 6 + + # Replicas have master IDs set + for i in range(3, 6): + assert self.nodes[i].masterid is not None + + # Write/read through cluster client + for i in range(50): + client.set(f"rkey:{i}", f"rval:{i}") + for i in range(50): + assert client.get(f"rkey:{i}") == f"rval:{i}".encode() + + @pytest.mark.order(2) + def test_setup_cluster_multiple_replicas_per_shard(self): + # 2 shards x (1 primary + 2 replicas) = 6 nodes + self.setup_cluster(num_shards=2, num_replicas_per_shard=2) + + assert len(self.nodes) == 6 + + # Cluster is healthy and every node sees all 6 + for node in self.nodes: + info = ClusterInfo(node.client.cluster("INFO")) + assert info.is_cluster_ok(), f"Node {node.port} not in ok state" + assert info.cluster_known_nodes() == 6 + + # Each primary has exactly 2 replicas online. + for i in range(2): + assert self.nodes[i].num_replicas_online() == 2 + + # Every replica is attached to one of the two primaries. + primary_ids = {self.nodes[0].nodeid, self.nodes[1].nodeid} + for i in range(2, 6): + assert self.nodes[i].masterid in primary_ids diff --git a/tests/test_slot_migration.py b/tests/test_slot_migration.py new file mode 100644 index 0000000..1cacac8 --- /dev/null +++ b/tests/test_slot_migration.py @@ -0,0 +1,152 @@ +""" +Slot Migration Tests + +Verifies ClusterTestCase.migrate_slot(): a slot (and the keys in it) moves from +one node to another on a live cluster, ownership updates across the cluster, and +the keys are served from the new owner afterward. Covers key movement (incl. +slots larger than one batch), multi-DB (requires cluster-databases > 1), +caller-DB isolation, cluster-wide ownership, empty-slot, and replica-shard +migrations. +""" + +import os + +from conftest import resource_port_tracker +from valkey_test_case import ClusterTestCase +from util.waiters import wait_for_equal +from valkey.cluster import key_slot + +version = os.environ.get("SERVER_VERSION", "unstable") +SERVER_PATH = os.path.join( + os.path.dirname(os.path.realpath(__file__)), + ".build", + "binaries", + version, + "valkey-server", +) + + +class TestSlotMigration(ClusterTestCase): + """Move slots between live nodes and confirm data follows.""" + + server_path = SERVER_PATH + + def _pick_source_and_target(self, slot): + source = self.get_slot_owner(slot) + target = next(n for n in self.nodes if n.nodeid != source.nodeid) + return source, target + + def test_migrate_slot_moves_keys(self): + """All keys in a slot move to the new owner, and their values survive.""" + self.setup_cluster(num_shards=2, num_replicas_per_shard=0) + + # Hash tag {m} forces every key into the same slot. Use enough keys to + # cross the 100-key GETKEYSINSLOT batch boundary migrate_slot drains. + slot = key_slot(b"{m}") + source, target = self._pick_source_and_target(slot) + for i in range(250): + source.client.set(f"{{m}}:{i}", i) + assert source.count_keys_in_slot(slot) == 250 + + self.migrate_slot(source, target, slot) + self.wait_for_slot_owner(slot, target) + + assert source.count_keys_in_slot(slot) == 0 + assert target.count_keys_in_slot(slot) == 250 + # A specific key's value survived the move, not just the count. + assert target.client.get("{m}:0") == b"0" + + def test_migrate_slot_multiple_databases(self): + """Keys sharing a slot across several DBs all move (cluster-databases).""" + self.args["cluster-databases"] = "16" + self.setup_cluster(num_shards=2, num_replicas_per_shard=0) + + key = "mdkey" + slot = key_slot(key.encode()) + source, target = self._pick_source_and_target(slot) + + dbs = (0, 1, 2) + for db in dbs: + source.create_from_server(db=db).set(key, f"v{db}") + + self.migrate_slot(source, target, slot, dbs=dbs) + self.wait_for_slot_owner(slot, target) + + for db in dbs: + assert target.create_from_server(db=db).get(key) == f"v{db}".encode() + + def test_migrate_does_not_disturb_caller_db(self): + """migrate_slot must not change the DB the caller's client is on.""" + self.args["cluster-databases"] = "16" + self.setup_cluster(num_shards=2, num_replicas_per_shard=0) + + key = "dbkey" + slot = key_slot(key.encode()) + source, target = self._pick_source_and_target(slot) + + # Put the source client on DB 3 before migrating. + source.client.execute_command("SELECT", 3) + source.client.set(key, "x") + + self.migrate_slot(source, target, slot, dbs=(3,)) + self.wait_for_slot_owner(slot, target) + + # The source client must still be on DB 3 (migrate used its own conn). + info = source.client.execute_command("CLIENT", "INFO") + info = info.decode() if isinstance(info, bytes) else info + assert " db=3 " in info, f"caller client left on wrong DB: {info}" + # And the key really did move to DB 3 on the target. + assert target.create_from_server(db=3).get(key) == b"x" + + def test_ownership_updates_across_cluster(self): + """Every node agrees the target owns the slot after migration.""" + self.setup_cluster(num_shards=2, num_replicas_per_shard=0) + + slot = key_slot(b"ownerkey") + source, target = self._pick_source_and_target(slot) + + self.migrate_slot(source, target, slot) + self.wait_for_slot_owner(slot, target) + + for node in self.nodes: + assert node.get_slot_owner_id(slot) == target.nodeid + + def test_migrate_empty_slot(self): + """Migrating a slot with no keys still transfers ownership cleanly.""" + self.setup_cluster(num_shards=2, num_replicas_per_shard=0) + + slot = key_slot(b"emptykey") + source, target = self._pick_source_and_target(slot) + assert source.count_keys_in_slot(slot) == 0 + + self.migrate_slot(source, target, slot) + self.wait_for_slot_owner(slot, target) + + assert self.get_slot_owner(slot).nodeid == target.nodeid + + def test_migrate_with_replicas(self): + """Migration works when shards have replicas. + + CLUSTER SETSLOT is only valid on primaries, so migrate_slot must skip + replica nodes; the migrated key must also reach the target's replica. + """ + self.setup_cluster(num_shards=2, num_replicas_per_shard=1) + + key = "replkey" + slot = key_slot(key.encode()) + source = self.get_slot_owner(slot) + # target = the other primary (a node that owns some slots, not source) + target = next( + n for n in self.nodes if n.is_primary() and n.nodeid != source.nodeid + ) + source.client.set(key, "payload") + + self.migrate_slot(source, target, slot) + self.wait_for_slot_owner(slot, target) + + assert target.client.get(key) == b"payload" + + # The key should replicate to the target primary's replica. + replica = next(n for n in self.nodes if n.masterid == target.nodeid) + replica.client.execute_command("READONLY") + wait_for_equal(lambda: replica.client.get(key), b"payload")