Add VSS Store Implementation

A KVStore implementation that writes to and reads
from a [VSS](https://github.com/lightningdevkit/vss-server/blob/main/README.md)
backend.

Author: G8XSU

.github/workflows/build.yml

@@ -48,6 +48,7 @@ jobs:
           cargo update -p regex --precise "1.9.6" --verbose # regex 1.10.0 requires rustc 1.65.0
           cargo update -p jobserver --precise "0.1.26" --verbose # jobserver 0.1.27 requires rustc 1.66.0
           cargo update -p zstd-sys --precise "2.0.8+zstd.1.5.5" --verbose # zstd-sys 2.0.9+zstd.1.5.5 requires rustc 1.64.0
+          cargo update -p petgraph --precise "0.6.3" --verbose # petgraph v0.6.4, requires rustc 1.64 or newer
       - name: Build on Rust ${{ matrix.toolchain }}
         run: cargo build --verbose --color always
       - name: Build with UniFFI support on Rust ${{ matrix.toolchain }}

Cargo.toml

@@ -66,6 +66,7 @@ tokio = { version = "1", default-features = false, features = [ "rt-multi-thread
 esplora-client = { version = "0.4", default-features = false }
 libc = "0.2"
 uniffi = { version = "0.23.0", features = ["build"], optional = true }
+vss-client = "0.1"
 
 [target.'cfg(windows)'.dependencies]
 winapi = { version = "0.3", features = ["winbase"] }

src/builder.rs

@@ -37,6 +37,8 @@ use lightning_persister::fs_store::FilesystemStore;
 
 use lightning_transaction_sync::EsploraSyncClient;
 
+#[cfg(any(vss, vss_test))]
+use crate::io::vss_store::VssStore;
 use bdk::bitcoin::secp256k1::Secp256k1;
 use bdk::blockchain::esplora::EsploraBlockchain;
 use bdk::database::SqliteDatabase;
@@ -269,6 +271,16 @@ impl NodeBuilder {
 		self.build_with_store(kv_store)
 	}
 
+	/// Builds a [`Node`] instance with a [`VssStore`] backend and according to the options
+	/// previously configured.
+	#[cfg(any(vss, vss_test))]
+	pub fn build_with_vss_store(
+		&self, url: &str, store_id: String,
+	) -> Result<Node<VssStore>, BuildError> {
+		let vss = Arc::new(VssStore::new(url, store_id));
+		self.build_with_store(vss)
+	}
+
 	/// Builds a [`Node`] instance according to the options previously configured.
 	pub fn build_with_store<K: KVStore + Sync + Send + 'static>(
 		&self, kv_store: Arc<K>,

src/io/mod.rs

@@ -4,6 +4,8 @@ pub mod sqlite_store;
 #[cfg(test)]
 pub(crate) mod test_utils;
 pub(crate) mod utils;
+#[cfg(any(vss, vss_test))]
+pub(crate) mod vss_store;
 
 /// The event queue will be persisted under this key.
 pub(crate) const EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE: &str = "";

src/io/vss_store.rs

@@ -0,0 +1,194 @@
+use io::Error;
+use std::io;
+use std::io::ErrorKind;
+#[cfg(test)]
+use std::panic::RefUnwindSafe;
+
+use crate::io::utils::check_namespace_key_validity;
+use lightning::util::persist::KVStore;
+use tokio::runtime::Runtime;
+use vss_client::client::VssClient;
+use vss_client::error::VssError;
+use vss_client::types::{
+	DeleteObjectRequest, GetObjectRequest, KeyValue, ListKeyVersionsRequest, PutObjectRequest,
+};
+
+/// A [`KVStore`] implementation that writes to and reads from a [VSS](https://github.com/lightningdevkit/vss-server/blob/main/README.md) backend.
+pub struct VssStore {
+	client: VssClient,
+	store_id: String,
+	runtime: Runtime,
+}
+
+impl VssStore {
+	pub(crate) fn new(base_url: &str, store_id: String) -> Self {
+		let client = VssClient::new(base_url);
+		let runtime = tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap();
+		Self { client, store_id, runtime }
+	}
+
+	fn build_key(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+	) -> io::Result<String> {
+		if primary_namespace.is_empty() {
+			Ok(key.to_string())
+		} else {
+			Ok(format!("{}#{}#{}", primary_namespace, secondary_namespace, key))
+		}
+	}
+
+	fn extract_key(&self, unified_key: &str) -> io::Result<String> {
+		let mut parts = unified_key.splitn(3, '#');
+		let (_primary_namespace, _secondary_namespace) = (parts.next(), parts.next());
+		match parts.next() {
+			Some(actual_key) => Ok(actual_key.to_string()),
+			None => Err(Error::new(ErrorKind::InvalidData, "Invalid key format")),
+		}
+	}
+
+	async fn list_all_keys(
+		&self, primary_namespace: &str, secondary_namespace: &str,
+	) -> io::Result<Vec<String>> {
+		let mut page_token = None;
+		let mut keys = vec![];
+		let key_prefix = format!("{}#{}", primary_namespace, secondary_namespace);
+		while page_token != Some("".to_string()) {
+			let request = ListKeyVersionsRequest {
+				store_id: self.store_id.clone(),
+				key_prefix: Some(key_prefix.clone()),
+				page_token,
+				page_size: None,
+			};
+
+			let response = self.client.list_key_versions(&request).await.map_err(|e| {
+				let msg = format!(
+					"Failed to list keys in {}/{}: {}",
+					primary_namespace, secondary_namespace, e
+				);
+				Error::new(ErrorKind::Other, msg)
+			})?;
+
+			for kv in response.key_versions {
+				keys.push(self.extract_key(&kv.key)?);
+			}
+			page_token = response.next_page_token;
+		}
+		Ok(keys)
+	}
+}
+
+impl KVStore for VssStore {
+	fn read(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+	) -> io::Result<Vec<u8>> {
+		check_namespace_key_validity(primary_namespace, secondary_namespace, Some(key), "read")?;
+		let request = GetObjectRequest {
+			store_id: self.store_id.clone(),
+			key: self.build_key(primary_namespace, secondary_namespace, key)?,
+		};
+
+		let resp =
+			tokio::task::block_in_place(|| self.runtime.block_on(self.client.get_object(&request)))
+				.map_err(|e| {
+					let msg = format!(
+						"Failed to read from key {}/{}/{}: {}",
+						primary_namespace, secondary_namespace, key, e
+					);
+					match e {
+						VssError::NoSuchKeyError(..) => Error::new(ErrorKind::NotFound, msg),
+						_ => Error::new(ErrorKind::Other, msg),
+					}
+				})?;
+		Ok(resp.value.unwrap().value)
+	}
+
+	fn write(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8],
+	) -> io::Result<()> {
+		check_namespace_key_validity(primary_namespace, secondary_namespace, Some(key), "write")?;
+		let request = PutObjectRequest {
+			store_id: self.store_id.clone(),
+			global_version: None,
+			transaction_items: vec![KeyValue {
+				key: self.build_key(primary_namespace, secondary_namespace, key)?,
+				version: -1,
+				value: buf.to_vec(),
+			}],
+			delete_items: vec![],
+		};
+
+		tokio::task::block_in_place(|| self.runtime.block_on(self.client.put_object(&request)))
+			.map_err(|e| {
+				let msg = format!(
+					"Failed to write to key {}/{}/{}: {}",
+					primary_namespace, secondary_namespace, key, e
+				);
+				Error::new(ErrorKind::Other, msg)
+			})?;
+
+		Ok(())
+	}
+
+	fn remove(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, _lazy: bool,
+	) -> io::Result<()> {
+		check_namespace_key_validity(primary_namespace, secondary_namespace, Some(key), "remove")?;
+		let request = DeleteObjectRequest {
+			store_id: self.store_id.clone(),
+			key_value: Some(KeyValue {
+				key: self.build_key(primary_namespace, secondary_namespace, key)?,
+				version: -1,
+				value: vec![],
+			}),
+		};
+
+		tokio::task::block_in_place(|| self.runtime.block_on(self.client.delete_object(&request)))
+			.map_err(|e| {
+				let msg = format!(
+					"Failed to delete key {}/{}/{}: {}",
+					primary_namespace, secondary_namespace, key, e
+				);
+				Error::new(ErrorKind::Other, msg)
+			})?;
+		Ok(())
+	}
+
+	fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> io::Result<Vec<String>> {
+		check_namespace_key_validity(primary_namespace, secondary_namespace, None, "list")?;
+
+		let keys = tokio::task::block_in_place(|| {
+			self.runtime.block_on(self.list_all_keys(primary_namespace, secondary_namespace))
+		})
+		.map_err(|e| {
+			let msg = format!(
+				"Failed to retrieve keys in namespace: {}/{} : {}",
+				primary_namespace, secondary_namespace, e
+			);
+			Error::new(ErrorKind::Other, msg)
+		})?;
+
+		Ok(keys)
+	}
+}
+
+#[cfg(test)]
+impl RefUnwindSafe for VssStore {}
+
+#[cfg(test)]
+#[cfg(vss_test)]
+mod tests {
+	use super::*;
+	use crate::io::test_utils::do_read_write_remove_list_persist;
+	use rand::distributions::Alphanumeric;
+	use rand::{thread_rng, Rng};
+
+	#[test]
+	fn read_write_remove_list_persist() {
+		let vss_base_url = std::env::var("TEST_VSS_BASE_URL").unwrap();
+		let mut rng = thread_rng();
+		let rand_store_id: String = (0..7).map(|_| rng.sample(Alphanumeric) as char).collect();
+		let vss_store = VssStore::new(&vss_base_url, rand_store_id);
+
+		do_read_write_remove_list_persist(&vss_store);
+	}
+}

src/test/functional_tests.rs

@@ -18,6 +18,29 @@ fn channel_full_cycle() {
 	do_channel_full_cycle(node_a, node_b, &bitcoind, &electrsd, false);
 }
 
+#[test]
+#[cfg(vss_test)]
+fn channel_full_cycle_with_vss_store() {
+	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();
+	println!("== Node A ==");
+	let esplora_url = format!("http://{}", electrsd.esplora_url.as_ref().unwrap());
+	let config_a = random_config();
+	let mut builder_a = NodeBuilder::from_config(config_a);
+	builder_a.set_esplora_server(esplora_url.clone());
+	let vss_base_url = std::env::var("TEST_VSS_BASE_URL").unwrap();
+	let node_a = builder_a.build_with_vss_store(&vss_base_url, "node_1_store".to_string()).unwrap();
+	node_a.start().unwrap();
+
+	println!("\n== Node B ==");
+	let config_b = random_config();
+	let mut builder_b = NodeBuilder::from_config(config_b);
+	builder_b.set_esplora_server(esplora_url);
+	let node_b = builder_b.build_with_vss_store(&vss_base_url, "node_2_store".to_string()).unwrap();
+	node_b.start().unwrap();
+
+	do_channel_full_cycle(node_a, node_b, &bitcoind, &electrsd, false);
+}
+
 #[test]
 fn channel_full_cycle_0conf() {
 	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();