diff --git a/src/async_kv_store.rs b/src/async_kv_store.rs index 1466e4f7..fb0eb021 100644 --- a/src/async_kv_store.rs +++ b/src/async_kv_store.rs @@ -255,13 +255,13 @@ pub struct BpKvStoreRouter { rest: Arc, } -enum BpRoute { +pub(crate) enum BpRoute { RemoteFirst, LocalOnly, Rest, } -fn bp_route(primary: &str, secondary: &str, key: &str) -> BpRoute { +pub(crate) fn bp_route(primary: &str, secondary: &str, key: &str) -> BpRoute { use lightning::util::persist::{ CHANNEL_MANAGER_PERSISTENCE_KEY, CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_KEY, diff --git a/src/kv_store.rs b/src/kv_store.rs index 6849a900..975e9f80 100644 --- a/src/kv_store.rs +++ b/src/kv_store.rs @@ -109,6 +109,16 @@ impl SeaOrmKvStore { &self.connection } + #[cfg(feature = "vss")] + pub(crate) fn list_all( + &self, + ) -> Result, io::Error> { + block_on(KvStoreEntity::find().all(self.get_connection())).map_err(|e| { + tracing::error!(error = %e, "KVStore list_all failed"); + io::Error::new(io::ErrorKind::Other, format!("Database list failed: {e}")) + }) + } + /// Atomically writes a local value and its durable VSS replication intent. /// /// A process may terminate immediately after this transaction commits. Keeping both rows in diff --git a/src/ldk.rs b/src/ldk.rs index 5fabb26c..aa52849b 100644 --- a/src/ldk.rs +++ b/src/ldk.rs @@ -150,8 +150,8 @@ const PENDING_FUNDING_NAMESPACE: &str = "pending_funding"; const FUNDING_CONSIGNMENT_NAMESPACE: &str = "funding_consignment"; /// Local-only marker: absent on a freshly restored device, so the fascia /// replay reruns until it completes once. -const REIMPORT_MARKER_NAMESPACE: &str = "reimport_marker"; -const REIMPORT_MARKER_KEY: &str = "fascia_replay"; +pub(crate) const REIMPORT_MARKER_NAMESPACE: &str = "reimport_marker"; +pub(crate) const REIMPORT_MARKER_KEY: &str = "fascia_replay"; const CONFIG_INDEXER_URL: &str = "indexer_url"; const CONFIG_BITCOIN_NETWORK: &str = "bitcoin_network"; const CONFIG_WALLET_FINGERPRINT: &str = "wallet_fingerprint"; @@ -4880,6 +4880,12 @@ pub(crate) async fn start_ldk( } } } + } else { + // Local state is authoritative: refill whatever the remote lacks + // (wiped or partial store) without overwriting what it holds. + synced.push_missing_to_vss().map_err(|e| { + APIError::FailedVssInit(format!("VSS resync of local state failed: {e}")) + })?; } (synced, monitor_kv_store) @@ -5461,6 +5467,20 @@ pub(crate) async fn start_ldk( )) })?; tracing::info!("VSS auto-backup (blocking) enabled for RGB wallet"); + // Auto-backup only tracks local changes; an empty remote (fresh + // wallet or wiped store) needs an explicit upload. + if let Some(client) = rgb_wallet.vss_client() { + let rt = client.handle().clone(); + let info = rt + .block_on(rgb_wallet.vss_backup_info(&client)) + .map_err(|e| APIError::FailedVssInit(format!("VSS backup info: {e}")))?; + if !info.backup_exists { + rt.block_on(rgb_wallet.vss_backup(&client)).map_err(|e| { + APIError::FailedVssInit(format!("initial RGB VSS backup failed: {e}")) + })?; + tracing::info!("Uploaded RGB wallet backup to empty VSS store"); + } + } } Ok::<_, APIError>((rgb_wallet, rgb_online)) }) diff --git a/src/synced_kv_store.rs b/src/synced_kv_store.rs index 6a3089dd..95b7cc33 100644 --- a/src/synced_kv_store.rs +++ b/src/synced_kv_store.rs @@ -405,6 +405,33 @@ impl SyncedKvStore { Ok(restored) } + /// Pushes local rows the remote lacks, never overwriting what it holds. + /// Refills a VSS store that was wiped or is otherwise incomplete. + #[cfg(feature = "vss")] + pub(crate) fn push_missing_to_vss(&self) -> Result { + let Some(ref remote) = self.remote else { + return Ok(0); + }; + let remote_keys: std::collections::HashSet = + remote.list_all_keys()?.into_iter().collect(); + let mut pushed = 0usize; + for row in self.local.list_all()? { + let (primary, secondary, key) = + (&row.primary_namespace, &row.secondary_namespace, &row.key); + if is_local_only(primary, secondary, key) + || remote_keys.contains(&crate::vss_kv_store::vss_key(primary, secondary, key)) + { + continue; + } + remote.write(primary, secondary, key, row.value)?; + pushed += 1; + } + if pushed > 0 { + tracing::info!(pushed, "Pushed local keys missing from VSS"); + } + Ok(pushed) + } + /// Local-only write; the row is never replicated, so a wipe-and-restore /// intentionally loses it. pub(crate) fn write_local_only( @@ -622,6 +649,15 @@ impl SyncedKvStore { } } +#[cfg(feature = "vss")] +fn is_local_only(primary: &str, secondary: &str, key: &str) -> bool { + use crate::async_kv_store::{bp_route, BpRoute}; + primary == PENDING_NS + || (primary == crate::ldk::REIMPORT_MARKER_NAMESPACE + && key == crate::ldk::REIMPORT_MARKER_KEY) + || matches!(bp_route(primary, secondary, key), BpRoute::LocalOnly) +} + impl KVStoreSync for SyncedKvStore { fn read( &self, diff --git a/src/test/vss.rs b/src/test/vss.rs index aa6eba62..fedce3e2 100644 --- a/src/test/vss.rs +++ b/src/test/vss.rs @@ -449,6 +449,96 @@ mod tests { assert_eq!(cm_keys, vec!["manager"]); } + /// A wiped VSS store is refilled from local state: keys missing on the + /// remote are pushed, keys the remote already holds are left untouched, + /// and local-only rows (pending queue, marker, graph/scorer) stay local. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn synced_kv_store_push_fills_missing_remote_keys() { + use lightning::util::persist::{ + NETWORK_GRAPH_PERSISTENCE_KEY, NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + }; + if !vss_server_available() { + eprintln!("SKIP: VSS server not available at {VSS_URL}"); + return; + } + + let (signing_key, store_id) = generate_test_keys(); + let db = create_test_sqlite(); + let local = Arc::new(SeaOrmKvStore::from_connection(db)); + let vss = Arc::new( + VssKvStore::new(VSS_URL.to_string(), store_id, signing_key).expect("vss store"), + ); + let synced = SyncedKvStore::with_vss(local.clone(), vss.clone()); + + synced + .write("channel_manager", "", "manager", vec![0xCA; 64]) + .unwrap(); + synced + .write("monitors", "", "deadbeef_0", vec![0xBE; 64]) + .unwrap(); + synced + .write("monitors", "", "deadbeef_1", vec![0xBF; 64]) + .unwrap(); + synced + .write_local_only("reimport_marker", "", "fascia_replay", vec![1]) + .unwrap(); + local + .write( + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, + NETWORK_GRAPH_PERSISTENCE_KEY, + vec![0x11; 8], + ) + .unwrap(); + local + .write( + crate::synced_kv_store::PENDING_NS, + "", + "x//y", + vec![1, 0xAA], + ) + .unwrap(); + + // Wipe the remote, then plant a foreign value the push must not clobber. + for k in vss.list_all_keys().expect("list keys") { + let (p, s, key) = parse_vss_key(&k).expect("rln key"); + vss.remove(&p, &s, &key, false).expect("remote remove"); + } + assert!(vss.list_all_keys().expect("list keys").is_empty()); + vss.write("monitors", "", "deadbeef_1", vec![0xEE; 64]) + .unwrap(); + + assert_eq!(synced.push_missing_to_vss().expect("push"), 2); + + assert_eq!( + vss.read("channel_manager", "", "manager").unwrap(), + vec![0xCA; 64] + ); + assert_eq!( + vss.read("monitors", "", "deadbeef_0").unwrap(), + vec![0xBE; 64] + ); + assert_eq!( + vss.read("monitors", "", "deadbeef_1").unwrap(), + vec![0xEE; 64], + "existing remote value must not be overwritten" + ); + let mut remote_keys = vss.list_all_keys().expect("list keys"); + remote_keys.sort(); + assert_eq!( + remote_keys, + vec![ + vss_key("channel_manager", "", "manager"), + vss_key("monitors", "", "deadbeef_0"), + vss_key("monitors", "", "deadbeef_1"), + ], + "local-only rows must not be pushed" + ); + + assert_eq!(synced.push_missing_to_vss().expect("push again"), 0); + } + /// Edge case: VSS server unreachable. Writes must still succeed locally /// (the local store is authoritative) and the failed replication must be /// queued for later retry. Does not require a running VSS server. @@ -895,6 +985,97 @@ mod tests { crate::test::shutdown(&[node_address]).await; } + /// A VSS store wiped while the node keeps its local state is refilled at + /// the next unlock: node keys are pushed and the RGB backup re-uploaded. + #[serial_test::serial] + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn vss_wiped_store_is_refilled_on_unlock() { + if !vss_server_available() { + eprintln!("SKIP: VSS server not available at {VSS_URL}"); + return; + } + tokio::time::timeout( + std::time::Duration::from_secs(180), + wiped_store_is_refilled_inner(), + ) + .await + .expect("vss_wiped_store_is_refilled_on_unlock timed out"); + } + + async fn wiped_store_is_refilled_inner() { + use lightning::util::persist::{ + CHANNEL_MANAGER_PERSISTENCE_KEY, CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + }; + use rgb_lib::bdk_wallet::keys::bip39::Mnemonic; + + crate::test::initialize(); + + let test_dir_node = "tmp/vss_wiped_store_refilled/node1"; + let node_address = crate::test::start_daemon_with_vss( + test_dir_node, + crate::test::NODE1_PEER_PORT, + false, + Some(VSS_URL.to_string()), + false, + ) + .await; + let password = "vss_wiped_store_refilled"; + let mnemonic = crate::test::init(node_address, password, None) + .await + .mnemonic; + unlock_with_electrum_backend(node_address, password).await; + crate::test::lock(node_address).await; + + let identity = crate::ldk::derive_vss_identity( + &Mnemonic::parse(&mnemonic).unwrap(), + bitcoin::Network::Regtest, + ) + .unwrap(); + let node_store = VssKvStore::new( + VSS_URL.to_string(), + identity.pubkey_hex.clone(), + identity.signing_key, + ) + .unwrap(); + let rgb_store = VssKvStore::new( + VSS_URL.to_string(), + format!("{}_rgb", identity.pubkey_hex), + identity.signing_key, + ) + .unwrap(); + + let before = node_store.list_all_keys().unwrap(); + assert!(before.contains(&vss_key( + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + CHANNEL_MANAGER_PERSISTENCE_KEY, + ))); + assert!(!rgb_store.list_all_keys().unwrap().is_empty()); + for store in [&node_store, &rgb_store] { + for key in store.list_all_keys().unwrap() { + store.remove_raw(&key).unwrap(); + } + assert!(store.list_all_keys().unwrap().is_empty()); + } + + unlock_with_electrum_backend(node_address, password).await; + + let after = node_store.list_all_keys().unwrap(); + for key in &before { + assert!(after.contains(key), "key missing after refill: {key}"); + } + let info: serde_json::Value = reqwest::get(format!("http://{node_address}/vssbackupinfo")) + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(info["backup_exists"], true, "{info}"); + + crate::test::shutdown(&[node_address]).await; + } + /// A graceful lock must release the VSS fence: the next unlock runs under /// a fresh instance id and must take over without `/vssclearfence`. #[serial_test::serial] diff --git a/src/vss_kv_store.rs b/src/vss_kv_store.rs index 108667d7..38a20fd5 100644 --- a/src/vss_kv_store.rs +++ b/src/vss_kv_store.rs @@ -625,6 +625,89 @@ impl VssKvStore { Ok(all_items) } + + /// Deletes a key by its raw VSS name, e.g. one of rgb-lib's backup keys. + pub(crate) fn remove_raw(&self, vss_key: &str) -> Result<(), io::Error> { + // VSS honors `version = -1` only for puts; `delete_items` require the + // object's current version, so a blind delete is rejected with a + // version conflict and, once queued, retries forever without ever + // converging. Read the current version and issue a conditional delete; + // an absent key means the removal goal is already met. + let get_req = GetObjectRequest { + store_id: self.store_id.clone(), + key: vss_key.to_string(), + }; + let existing_version = match self.block_on(self.client.get_object(&get_req)) { + Ok(resp) => match resp.value { + Some(kv) => kv.version, + None => return Ok(()), + }, + Err(VssError::NoSuchKeyError(_)) => return Ok(()), + Err(e) => { + tracing::error!(vss_key, error = %e, "VssKvStore remove read failed"); + return Err(io::Error::new( + io::ErrorKind::Other, + format!("VSS remove read failed: {e}"), + )); + } + }; + + let request = PutObjectRequest { + store_id: self.store_id.clone(), + global_version: None, + transaction_items: vec![], + delete_items: vec![KeyValue { + key: vss_key.to_string(), + version: existing_version, + value: vec![], + }], + }; + + match self.block_on(self.client.put_object(&request)) { + Ok(_) | Err(VssError::NoSuchKeyError(_)) => Ok(()), + Err(e) => { + tracing::error!(vss_key, error = %e, "VssKvStore remove failed"); + Err(io::Error::new( + io::ErrorKind::Other, + format!("VSS remove failed: {e}"), + )) + } + } + } + + /// Lists every key in the store except the fence. + pub fn list_all_keys(&self) -> Result, io::Error> { + let mut keys = Vec::new(); + let mut page_token: Option = None; + loop { + let list_req = ListKeyVersionsRequest { + store_id: self.store_id.clone(), + key_prefix: None, + page_size: None, + page_token: page_token.clone(), + }; + let response = self + .block_on(self.client.list_key_versions(&list_req)) + .map_err(|e| { + io::Error::new( + io::ErrorKind::Other, + format!("VSS list_key_versions failed: {e}"), + ) + })?; + keys.extend( + response + .key_versions + .into_iter() + .map(|kv| kv.key) + .filter(|k| k != FENCE_KEY), + ); + match response.next_page_token { + Some(token) if !token.is_empty() => page_token = Some(token), + _ => break, + } + } + Ok(keys) + } } /// Encode a `(primary_namespace, secondary_namespace, key)` triple as a @@ -770,52 +853,7 @@ impl KVStoreSync for VssKvStore { tracing::trace!(vss_key, "VssKvStore remove"); self.check_fence_periodic(); - - // VSS honors `version = -1` only for puts; `delete_items` require the - // object's current version, so a blind delete is rejected with a - // version conflict and, once queued, retries forever without ever - // converging. Read the current version and issue a conditional delete; - // an absent key means the removal goal is already met. - let get_req = GetObjectRequest { - store_id: self.store_id.clone(), - key: vss_key.clone(), - }; - let existing_version = match self.block_on(self.client.get_object(&get_req)) { - Ok(resp) => match resp.value { - Some(kv) => kv.version, - None => return Ok(()), - }, - Err(VssError::NoSuchKeyError(_)) => return Ok(()), - Err(e) => { - tracing::error!(vss_key, error = %e, "VssKvStore remove read failed"); - return Err(io::Error::new( - io::ErrorKind::Other, - format!("VSS remove read failed: {e}"), - )); - } - }; - - let request = PutObjectRequest { - store_id: self.store_id.clone(), - global_version: None, - transaction_items: vec![], - delete_items: vec![KeyValue { - key: vss_key.clone(), - version: existing_version, - value: vec![], - }], - }; - - match self.block_on(self.client.put_object(&request)) { - Ok(_) | Err(VssError::NoSuchKeyError(_)) => Ok(()), - Err(e) => { - tracing::error!(vss_key, error = %e, "VssKvStore remove failed"); - Err(io::Error::new( - io::ErrorKind::Other, - format!("VSS remove failed: {e}"), - )) - } - } + self.remove_raw(&vss_key) } fn list(