store sync tests

2026-01-03 14:33:07 +00:00 · 2024-09-10 09:59:01 +03:00 · 2024-09-10 09:59:01 +03:00 · d3312538a7
commit d3312538a7
parent 437f6c0826
1 changed files with 169 additions and 0 deletions
--- a/tests/store_sync/test_store_sync.py
+++ b/tests/store_sync/test_store_sync.py
@ -264,3 +264,172 @@ class TestStoreSync(StepsStore):
        assert (
            node1_message == node2_message == node3_message == self.num_messages
        ), f"Store messages are not equal to each other or not equal to {self.num_messages}"
+
+    def test_sync_with_nodes_restart__case1(self):
+        self.node1.start(store="true", store_sync="true", relay="true")
+        self.node2.start(
+            store="true",
+            store_sync="true",
+            relay="true",
+            storenode=self.node1.get_multiaddr_with_id(),
+            discv5_bootstrap_node=self.node1.get_enr_uri(),
+        )
+        self.node3.start(
+            store="false",
+            store_sync="true",
+            relay="true",
+            storenode=self.node2.get_multiaddr_with_id(),
+            discv5_bootstrap_node=self.node2.get_enr_uri(),
+        )
+
+        self.add_node_peer(self.node2, [self.node1.get_multiaddr_with_id()])
+        self.add_node_peer(self.node3, [self.node2.get_multiaddr_with_id()])
+
+        self.node1.set_relay_subscriptions([self.test_pubsub_topic])
+        self.node2.set_relay_subscriptions([self.test_pubsub_topic])
+        self.node3.set_relay_subscriptions([self.test_pubsub_topic])
+
+        ml1 = [self.publish_message(sender=self.node1, via="relay", message_propagation_delay=0.01) for _ in range(self.num_messages)]
+        ml2 = [self.publish_message(sender=self.node2, via="relay", message_propagation_delay=0.01) for _ in range(self.num_messages)]
+        ml3 = [self.publish_message(sender=self.node3, via="relay", message_propagation_delay=0.01) for _ in range(self.num_messages)]
+
+        self.node1.restart()
+        self.node2.restart()
+        self.node3.restart()
+
+        delay(2)  # wait for the sync to finish
+
+        self.check_published_message_is_stored(page_size=100, ascending="true", store_node=self.node1, messages_to_check=ml1 + ml2 + ml3)
+        node1_message = len(self.store_response.messages)
+        self.check_published_message_is_stored(page_size=100, ascending="true", store_node=self.node2, messages_to_check=ml1 + ml2 + ml3)
+        node2_message = len(self.store_response.messages)
+        self.check_published_message_is_stored(page_size=100, ascending="true", store_node=self.node3, messages_to_check=ml1 + ml2 + ml3)
+        node3_message = len(self.store_response.messages)
+
+        assert (
+            node1_message == node2_message == node3_message == self.num_messages * 3
+        ), f"Store messages are not equal to each other or not equal to {self.num_messages * 3}"
+
+    def test_sync_with_nodes_restart__case2(self):
+        self.node1.start(store="true", relay="true")
+        self.node2.start(store="false", store_sync="true", relay="true", discv5_bootstrap_node=self.node1.get_enr_uri())
+        self.node3.start(store="false", store_sync="true", relay="true", discv5_bootstrap_node=self.node2.get_enr_uri())
+
+        self.add_node_peer(self.node2, [self.node1.get_multiaddr_with_id()])
+        self.add_node_peer(self.node3, [self.node2.get_multiaddr_with_id()])
+
+        self.node1.set_relay_subscriptions([self.test_pubsub_topic])
+        self.node2.set_relay_subscriptions([self.test_pubsub_topic])
+        self.node3.set_relay_subscriptions([self.test_pubsub_topic])
+
+        ml1 = [self.publish_message(sender=self.node1, via="relay", message_propagation_delay=0.01) for _ in range(self.num_messages)]
+        ml2 = [self.publish_message(sender=self.node2, via="relay", message_propagation_delay=0.01) for _ in range(self.num_messages)]
+        ml3 = [self.publish_message(sender=self.node3, via="relay", message_propagation_delay=0.01) for _ in range(self.num_messages)]
+
+        self.node2.restart()
+
+        delay(5)  # wait for the sync to finish
+
+        self.check_published_message_is_stored(page_size=100, ascending="true", store_node=self.node1, messages_to_check=ml1 + ml2 + ml3)
+        node1_message = len(self.store_response.messages)
+        self.check_published_message_is_stored(page_size=100, ascending="true", store_node=self.node2, messages_to_check=ml1 + ml2 + ml3)
+        node2_message = len(self.store_response.messages)
+        self.check_published_message_is_stored(page_size=100, ascending="true", store_node=self.node3, messages_to_check=ml1 + ml2 + ml3)
+        node3_message = len(self.store_response.messages)
+
+        assert (
+            node1_message == node2_message == node3_message == self.num_messages * 3
+        ), f"Store messages are not equal to each other or not equal to {self.num_messages * 3}"
+
+    def test_high_message_volume_sync(self):
+        self.node1.start(store="true", store_sync="true", relay="true")
+        self.node2.start(
+            store="true",
+            store_sync="true",
+            relay="true",
+            storenode=self.node1.get_multiaddr_with_id(),
+            discv5_bootstrap_node=self.node1.get_enr_uri(),
+        )
+        self.node3.start(
+            store="false",
+            store_sync="true",
+            relay="true",
+            storenode=self.node2.get_multiaddr_with_id(),
+            discv5_bootstrap_node=self.node2.get_enr_uri(),
+        )
+
+        self.add_node_peer(self.node2, [self.node1.get_multiaddr_with_id()])
+        self.add_node_peer(self.node3, [self.node2.get_multiaddr_with_id()])
+
+        self.node1.set_relay_subscriptions([self.test_pubsub_topic])
+        self.node2.set_relay_subscriptions([self.test_pubsub_topic])
+        self.node3.set_relay_subscriptions([self.test_pubsub_topic])
+
+        expected_message_hash_list = []
+
+        for _ in range(500):  # total 1500 messages
+            messages = [self.create_message() for _ in range(3)]
+
+            for i, node in enumerate([self.node1, self.node2, self.node3]):
+                self.publish_message(sender=node, via="relay", message=messages[i], message_propagation_delay=0.01)
+
+            expected_message_hash_list.extend([self.compute_message_hash(self.test_pubsub_topic, msg) for msg in messages])
+
+        delay(5)  # wait for the sync to finish
+
+        for node in [self.node1, self.node2, self.node3]:
+            store_response = StoreResponse({"paginationCursor": "", "pagination_cursor": ""}, node)
+            response_message_hash_list = []
+            while store_response.pagination_cursor is not None:
+                cursor = store_response.pagination_cursor
+                store_response = self.get_messages_from_store(node, page_size=100, cursor=cursor)
+                for index in range(len(store_response.messages)):
+                    response_message_hash_list.append(store_response.message_hash(index))
+            assert len(expected_message_hash_list) == len(response_message_hash_list), "Message count mismatch"
+            assert expected_message_hash_list == response_message_hash_list, "Message hash mismatch"
+
+    def test_large_message_payload_sync(self):
+        self.node1.start(store="true", relay="true")
+        self.node2.start(store="false", store_sync="true", relay="true", discv5_bootstrap_node=self.node1.get_enr_uri())
+        self.node3.start(store="false", store_sync="true", relay="true", discv5_bootstrap_node=self.node2.get_enr_uri())
+
+        self.add_node_peer(self.node2, [self.node1.get_multiaddr_with_id()])
+        self.add_node_peer(self.node3, [self.node2.get_multiaddr_with_id()])
+
+        self.node1.set_relay_subscriptions([self.test_pubsub_topic])
+        self.node2.set_relay_subscriptions([self.test_pubsub_topic])
+        self.node3.set_relay_subscriptions([self.test_pubsub_topic])
+
+        payload_length = 1024 * 100  # after encoding to base64 this will be close to 150KB
+
+        ml1 = [
+            self.publish_message(
+                sender=self.node1, via="relay", message=self.create_message(payload=to_base64("a" * (payload_length))), message_propagation_delay=0.01
+            )
+            for _ in range(self.num_messages)
+        ]
+        ml2 = [
+            self.publish_message(
+                sender=self.node2, via="relay", message=self.create_message(payload=to_base64("a" * (payload_length))), message_propagation_delay=0.01
+            )
+            for _ in range(self.num_messages)
+        ]
+        ml3 = [
+            self.publish_message(
+                sender=self.node3, via="relay", message=self.create_message(payload=to_base64("a" * (payload_length))), message_propagation_delay=0.01
+            )
+            for _ in range(self.num_messages)
+        ]
+
+        delay(10)  # wait for the sync to finish
+
+        self.check_published_message_is_stored(page_size=100, ascending="true", store_node=self.node1, messages_to_check=ml1 + ml2 + ml3)
+        node1_message = len(self.store_response.messages)
+        self.check_published_message_is_stored(page_size=100, ascending="true", store_node=self.node2, messages_to_check=ml1 + ml2 + ml3)
+        node2_message = len(self.store_response.messages)
+        self.check_published_message_is_stored(page_size=100, ascending="true", store_node=self.node3, messages_to_check=ml1 + ml2 + ml3)
+        node3_message = len(self.store_response.messages)
+
+        assert (
+            node1_message == node2_message == node3_message == self.num_messages * 3
+        ), f"Store messages are not equal to each other or not equal to {self.num_messages * 3}"