fixup! Add integration and FFI tests for tiered storage

Python test fix and update

Author: enigbe

bindings/python/src/ldk_node/kv_store.py

@@ -57,28 +57,23 @@ def dump(self):
 
     def read(self, primary_namespace: str, secondary_namespace: str, key: str) -> List[int]:
         with self._lock:
-            print(f"[{self.name}] READ: {primary_namespace}/{secondary_namespace}/{key}")
             namespace_key = (primary_namespace, secondary_namespace)
-            
+
             if namespace_key not in self.storage:
-                print(f"  -> namespace not found, keys: {list(self.storage.keys())}")
                 raise IoError.NotFound()
-            
+
             if key not in self.storage[namespace_key]:
-                print(f"  -> key not found, keys: {list(self.storage[namespace_key].keys())}")
                 raise IoError.NotFound()
 
-            data = self.storage[namespace_key][key]
-            print(f"  -> returning {len(data)} bytes")
-            return data
+            return list(self.storage[namespace_key][key])
 
     def write(self, primary_namespace: str, secondary_namespace: str, key: str, buf: List[int]) -> None:
         with self._lock:
             namespace_key = (primary_namespace, secondary_namespace)
             if namespace_key not in self.storage:
                 self.storage[namespace_key] = {}
 
-            self.storage[namespace_key][key] = buf
+            self.storage[namespace_key][key] = list(buf)
 
     def remove(self, primary_namespace: str, secondary_namespace: str, key: str, lazy: bool) -> None:
         with self._lock:
@@ -98,7 +93,7 @@ def list(self, primary_namespace: str, secondary_namespace: str) -> List[str]:
         with self._lock:
             namespace_key = (primary_namespace, secondary_namespace)
             if namespace_key in self.storage:
-                return list(self.storage[namespace_key].keys())
+                return sorted(self.storage[namespace_key].keys())
             return []
 
     async def read_async(self, primary_namespace: str, secondary_namespace: str, key: str) -> List[int]:

bindings/python/src/ldk_node/test_ldk_node.py

@@ -174,10 +174,10 @@ def setup_node_with_tier_store(tmp_dir, esplora_endpoint, listening_addresses) -
     builder.set_chain_source_esplora(esplora_endpoint, None)
     builder.set_network(DEFAULT_TEST_NETWORK)
     builder.set_listening_addresses(listening_addresses)
-    builder.set_backup_store(FfiDynStore.from_store(backup))
-    builder.set_ephemeral_store(FfiDynStore.from_store(ephemeral))
+    builder.set_backup_store(backup)
+    builder.set_ephemeral_store(ephemeral)
 
-    return builder.build_with_store(node_entropy, FfiDynStore.from_store(primary)), (primary, backup, ephemeral)
+    return builder.build_with_store(node_entropy, primary), (primary, backup, ephemeral)
 
 def do_channel_full_cycle(setup_1: NodeSetup, setup_2: NodeSetup, esplora_endpoint):
     # Fund both nodes
@@ -269,6 +269,32 @@ def expect_event(node, expected_event_type):
     node.event_handled()
     return event 
 
+def wait_until(predicate, timeout=10.0, interval=0.1, message="condition not met before timeout"):
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if predicate():
+            return
+        time.sleep(interval)
+    raise AssertionError(message)
+
+
+def stores_match(store_a, store_b):
+    if set(store_a.storage.keys()) != set(store_b.storage.keys()):
+        return False
+
+    for namespace_key in store_a.storage:
+        keys_a = store_a.storage[namespace_key]
+        keys_b = store_b.storage[namespace_key]
+
+        if set(keys_a.keys()) != set(keys_b.keys()):
+            return False
+
+        for key in keys_a:
+            if keys_a[key] != keys_b[key]:
+                return False
+
+    return True
+
 class TestLdkNode(unittest.TestCase):
     def setUp(self):
         bitcoin_cli("createwallet ldk_node_test")
@@ -287,40 +313,67 @@ def test_channel_full_cycle(self):
         setup_2.cleanup()
 
     def test_tier_store(self):
-        # Set event loop for async Python callbacks from Rust
+        # Set an event loop for async Python callbacks invoked from Rust.
         # (https://mozilla.github.io/uniffi-rs/0.27/futures.html#python-uniffi_set_event_loop)
         loop = asyncio.new_event_loop()
-        
+
         def run_loop():
             asyncio.set_event_loop(loop)
             loop.run_forever()
-        
+
         loop_thread = threading.Thread(target=run_loop, daemon=True)
         loop_thread.start()
         ldk_node.uniffi_set_event_loop(loop)
-        
+
         esplora_endpoint = get_esplora_endpoint()
-        setup_1, setup_2 = setup_two_nodes(esplora_endpoint, port_1=2325, port_2=2326, use_tier_store=True)
-        
-        do_channel_full_cycle(setup_1, setup_2, esplora_endpoint)
-        
-        primary, backup, ephemeral = setup_1.stores
-        
-        # Wait for async backup
-        time.sleep(2)  
-        
-        self.assertGreater(len(primary.storage), 0, "Primary should have data")
-        self.assertGreater(len(backup.storage), 0, "Backup should have data")
-        self.assertEqual(list(primary.storage.keys()), list(backup.storage.keys()), 
-                        "Backup should mirror primary")
-        
-        self.assertGreater(len(ephemeral.storage), 0, "Ephemeral should have data")
-        ephemeral_keys = [key for namespace in ephemeral.storage.values() for key in namespace.keys()]
-        has_scorer_or_graph = any(key in ['scorer', 'network_graph'] for key in ephemeral_keys)
-        self.assertTrue(has_scorer_or_graph, "Ephemeral should contain scorer or network_graph data")
-        
-        setup_1.cleanup()
-        setup_2.cleanup()
+        setup_1, setup_2 = setup_two_nodes(
+            esplora_endpoint,
+            port_1=2325,
+            port_2=2326,
+            use_tier_store=True,
+        )
+
+        try:
+            do_channel_full_cycle(setup_1, setup_2, esplora_endpoint)
+
+            primary, backup, ephemeral = setup_1.stores
+
+            wait_until(
+                lambda: len(primary.storage) > 0,
+                message="primary store did not receive any data",
+            )
+            wait_until(
+                lambda: len(backup.storage) > 0,
+                message="backup store did not receive any data",
+            )
+            wait_until(
+                lambda: stores_match(primary, backup),
+                timeout=15.0,
+                message="backup store did not converge to primary contents",
+            )
+
+            self.assertTrue(stores_match(primary, backup), "backup should mirror primary contents")
+
+            self.assertGreater(len(ephemeral.storage), 0, "ephemeral should have data")
+
+            ephemeral_keys = [
+                key
+                for namespace in ephemeral.storage.values()
+                for key in namespace.keys()
+            ]
+            has_scorer_or_graph = any(
+                key in ["scorer", "network_graph"] for key in ephemeral_keys
+            )
+            self.assertTrue(
+                has_scorer_or_graph,
+                "ephemeral should contain scorer or network_graph data",
+            )
+        finally:
+            setup_1.cleanup()
+            setup_2.cleanup()
+            loop.call_soon_threadsafe(loop.stop)
+            loop_thread.join(timeout=5)
+            loop.close()
 
 if __name__ == '__main__':
     unittest.main()