feat: tiered data storage

Cargo.toml

@@ -63,6 +63,7 @@ bitcoin = "0.32.7"
 bip39 = { version = "2.0.0", features = ["rand"] }
 bip21 = { version = "0.5", features = ["std"], default-features = false }
 
+async-trait = {version = "0.1.89"}
 base64 = { version = "0.22.1", default-features = false, features = ["std"] }
 rand = { version = "0.9.2", default-features = false, features = ["std", "thread_rng", "os_rng"] }
 chrono = { version = "0.4", default-features = false, features = ["clock"] }

benches/payments.rs

@@ -127,6 +127,7 @@ fn payment_benchmark(c: &mut Criterion) {
 		true,
 		false,
 		common::TestStoreType::Sqlite,
+		common::TestStoreType::Sqlite,
 	);
 
 	let runtime =

bindings/ldk_node.udl

@@ -69,6 +69,12 @@ enum WordCount {
 	"Words24",
 };
 
+dictionary RetryConfig {
+	u16 initial_retry_delay_ms;
+	u16 maximum_delay_ms;
+	f32 backoff_multiplier;
+};
+
 enum LogLevel {
 	"Gossip",
 	"Trace",
@@ -90,6 +96,56 @@ interface LogWriter {
 	void log(LogRecord record);
 };
 
+interface DynStore {
+    [Name=from_store]
+    constructor(SyncAndAsyncKVStore store);
+};
+
+[Trait, WithForeign]
+interface SyncAndAsyncKVStore {
+    // KVStoreSync versions
+    [Throws=IOError]
+    sequence<u8> read_sync(string primary_namespace, string secondary_namespace, string key);
+    [Throws=IOError]
+    void write_sync(string primary_namespace, string secondary_namespace, string key, sequence<u8> buf);
+    [Throws=IOError]
+    void remove_sync(string primary_namespace, string secondary_namespace, string key, boolean lazy);
+    [Throws=IOError]
+    sequence<string> list_sync(string primary_namespace, string secondary_namespace);
+
+    // KVStore versions
+    [Throws=IOError, Async]
+    sequence<u8> read_async(string primary_namespace, string secondary_namespace, string key);
+    [Throws=IOError, Async]
+    void write_async(string primary_namespace, string secondary_namespace, string key, sequence<u8> buf);
+    [Throws=IOError, Async]
+    void remove_async(string primary_namespace, string secondary_namespace, string key,  boolean lazy);
+    [Throws=IOError, Async]
+    sequence<string> list_async(string primary_namespace, string secondary_namespace);
+};
+
+[Error]
+enum IOError {
+  	"NotFound",
+    "PermissionDenied",
+    "ConnectionRefused",
+    "ConnectionReset",
+    "ConnectionAborted",
+    "NotConnected",
+    "AddrInUse",
+    "AddrNotAvailable",
+    "BrokenPipe",
+    "AlreadyExists",
+    "WouldBlock",
+    "InvalidInput",
+    "InvalidData",
+    "TimedOut",
+    "WriteZero",
+    "Interrupted",
+    "UnexpectedEof",
+    "Other",
+};
+
 interface Builder {
 	constructor();
 	[Name=from_config]
@@ -114,6 +170,9 @@ interface Builder {
 	void set_announcement_addresses(sequence<SocketAddress> announcement_addresses);
 	[Throws=BuildError]
 	void set_node_alias(string node_alias);
+	void set_tier_store_retry_config(RetryConfig retry_config);
+	void set_tier_store_backup(DynStore backup_store);
+	void set_tier_store_ephemeral(DynStore ephemeral_store);
 	[Throws=BuildError]
 	void set_async_payments_role(AsyncPaymentsRole? role);
 	[Throws=BuildError]
@@ -126,6 +185,8 @@ interface Builder {
 	Node build_with_vss_store_and_fixed_headers(NodeEntropy node_entropy, string vss_url, string store_id, record<string, string> fixed_headers);
 	[Throws=BuildError]
 	Node build_with_vss_store_and_header_provider(NodeEntropy node_entropy, string vss_url, string store_id, VssHeaderProvider header_provider);
+	[Throws=BuildError]
+	Node build_with_tier_store(NodeEntropy node_entropy, DynStore primary_store);
 };
 
 interface Node {

bindings/python/src/ldk_node/kv_store.py

@@ -0,0 +1,118 @@
+import threading
+
+from abc import ABC, abstractmethod
+from typing import List
+
+from ldk_node import IoError
+
+class AbstractKvStore(ABC):
+    @abstractmethod
+    async def list_async(self, primary_namespace: "str",secondary_namespace: "str") -> "typing.List[str]":
+        pass
+
+    @abstractmethod
+    def list_sync(self, primary_namespace: "str",secondary_namespace: "str") -> "typing.List[str]":
+        pass
+
+    @abstractmethod
+    async def read_async(self, primary_namespace: "str",secondary_namespace: "str",key: "str") -> "typing.List[int]":
+        pass
+
+    @abstractmethod
+    def read_sync(self, primary_namespace: "str",secondary_namespace: "str",key: "str") -> "typing.List[int]":
+        pass
+
+    @abstractmethod
+    async def remove_async(self, primary_namespace: "str",secondary_namespace: "str",key: "str",lazy: "bool") -> None:
+        pass
+
+    @abstractmethod
+    def remove_sync(self, primary_namespace: "str",secondary_namespace: "str",key: "str",lazy: "bool") -> None:
+        pass
+
+    @abstractmethod
+    async def write_async(self, primary_namespace: "str",secondary_namespace: "str",key: "str",buf: "typing.List[int]") -> None:
+        pass
+
+    @abstractmethod
+    def write_sync(self, primary_namespace: "str",secondary_namespace: "str",key: "str",buf: "typing.List[int]") -> None:
+        pass
+
+class TestKvStore(AbstractKvStore):
+    def __init__(self, name: str):
+        self.name = name
+        # Storage structure: {(primary_ns, secondary_ns): {key: [bytes]}}
+        self.storage = {}
+        self._lock = threading.Lock()
+
+    def dump(self):
+        print(f"\n[{self.name}] Store contents:")
+        for (primary_ns, secondary_ns), keys_dict in self.storage.items():
+            print(f"  Namespace: ({primary_ns!r}, {secondary_ns!r})")
+            for key, data in keys_dict.items():
+                print(f"    Key: {key!r} -> {len(data)} bytes")
+                # Optionally show first few bytes
+                preview = data[:20] if len(data) > 20 else data
+                print(f"      Data preview: {preview}...")
+
+    # KVStoreSync methods
+    def list_sync(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 []
+
+    def read_sync(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(f"Namespace not found: {primary_namespace}/{secondary_namespace}")
+
+            if key not in self.storage[namespace_key]:
+                print(f"  -> key not found, keys: {list(self.storage[namespace_key].keys())}")
+                raise IoError.NotFound(f"Key not found: {key}")
+
+            data = self.storage[namespace_key][key]
+            print(f"  -> returning {len(data)} bytes")
+            return data
+
+    def write_sync(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.copy()
+
+    def remove_sync(self, primary_namespace: str, secondary_namespace: str, key: str, lazy: bool) -> None:
+        with self._lock:
+            namespace_key = (primary_namespace, secondary_namespace)
+            if namespace_key not in self.storage:
+                raise IoError.NotFound(f"Namespace not found: {primary_namespace}/{secondary_namespace}")
+
+            if key not in self.storage[namespace_key]:
+                raise IoError.NotFound(f"Key not found: {key}")
+
+            del self.storage[namespace_key][key]
+
+            if not self.storage[namespace_key]:
+                del self.storage[namespace_key]
+
+    # KVStore methods
+    async def list_async(self, primary_namespace: str, secondary_namespace: str) -> List[str]:
+        return self.list_sync(primary_namespace, secondary_namespace)
+
+    async def read_async(self, primary_namespace: str, secondary_namespace: str, key: str) -> List[int]:
+        return self.read_sync(primary_namespace, secondary_namespace, key)
+
+    async def write_async(self, primary_namespace: str, secondary_namespace: str, key: str, buf: List[int]) -> None:
+        self.write_sync(primary_namespace, secondary_namespace, key, buf)
+
+    async def remove_async(self, primary_namespace: str, secondary_namespace: str, key: str, lazy: bool) -> None:
+        self.remove_sync(primary_namespace, secondary_namespace, key, lazy)
+
+