Merge pull request #651 from tnull/2025-09-fix-lazy-remove

tnull · web-flow · commit 98d21483c1f9 · 2025-10-10T09:23:04.000+02:00
Account for `Arc`s and `lazy` being dropped
diff --git a/Cargo.toml b/Cargo.toml
@@ -52,29 +52,29 @@ default = []
 #lightning-liquidity = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main" }
 #lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main" }
 
-lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204", features = ["std"] }
-lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204" }
-lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204", features = ["std"] }
-lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204" }
-lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204", features = ["tokio"] }
-lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204" }
-lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204" }
-lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204", features = ["rest-client", "rpc-client", "tokio"] }
-lightning-transaction-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204", features = ["esplora-async-https", "electrum-rustls-ring", "time"] }
-lightning-liquidity = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204" }
-lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204" }
-
-#lightning = { path = "../rust-lightning/lightning", features = ["std"] }
-#lightning-types = { path = "../rust-lightning/lightning-types" }
-#lightning-invoice = { path = "../rust-lightning/lightning-invoice", features = ["std"] }
-#lightning-net-tokio = { path = "../rust-lightning/lightning-net-tokio" }
-#lightning-persister = { path = "../rust-lightning/lightning-persister", features = ["tokio"] }
-#lightning-background-processor = { path = "../rust-lightning/lightning-background-processor" }
-#lightning-rapid-gossip-sync = { path = "../rust-lightning/lightning-rapid-gossip-sync" }
-#lightning-block-sync = { path = "../rust-lightning/lightning-block-sync", features = ["rest-client", "rpc-client", "tokio"] }
-#lightning-transaction-sync = { path = "../rust-lightning/lightning-transaction-sync", features = ["esplora-async-https", "electrum-rustls-ring", "time"] }
-#lightning-liquidity = { path = "../rust-lightning/lightning-liquidity", features = ["std"] }
-#lightning-macros = { path = "../rust-lightning/lightning-macros" }
+#lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204", features = ["std"] }
+#lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204" }
+#lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204", features = ["std"] }
+#lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204" }
+#lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204", features = ["tokio"] }
+#lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204" }
+#lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204" }
+#lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204", features = ["rest-client", "rpc-client", "tokio"] }
+#lightning-transaction-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204", features = ["esplora-async-https", "electrum-rustls-ring", "time"] }
+#lightning-liquidity = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204" }
+#lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204" }
+
+lightning = { path = "../rust-lightning/lightning", features = ["std"] }
+lightning-types = { path = "../rust-lightning/lightning-types" }
+lightning-invoice = { path = "../rust-lightning/lightning-invoice", features = ["std"] }
+lightning-net-tokio = { path = "../rust-lightning/lightning-net-tokio" }
+lightning-persister = { path = "../rust-lightning/lightning-persister", features = ["tokio"] }
+lightning-background-processor = { path = "../rust-lightning/lightning-background-processor" }
+lightning-rapid-gossip-sync = { path = "../rust-lightning/lightning-rapid-gossip-sync" }
+lightning-block-sync = { path = "../rust-lightning/lightning-block-sync", features = ["rest-client", "rpc-client", "tokio"] }
+lightning-transaction-sync = { path = "../rust-lightning/lightning-transaction-sync", features = ["esplora-async-https", "electrum-rustls-ring", "time"] }
+lightning-liquidity = { path = "../rust-lightning/lightning-liquidity", features = ["std"] }
+lightning-macros = { path = "../rust-lightning/lightning-macros" }
 
 bdk_chain = { version = "0.23.0", default-features = false, features = ["std"] }
 bdk_esplora = { version = "0.22.0", default-features = false, features = ["async-https-rustls", "tokio"]}
@@ -109,8 +109,8 @@ winapi = { version = "0.3", features = ["winbase"] }
 [dev-dependencies]
 #lightning = { version = "0.1.0", features = ["std", "_test_utils"] }
 #lightning = { git = "https://github.com/lightningdevkit/rust-lightning", branch="main", features = ["std", "_test_utils"] }
-lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204", features = ["std", "_test_utils"] }
-#lightning = { path = "../rust-lightning/lightning", features = ["std", "_test_utils"] }
+#lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "d0765847c85f1c3dc753c17c3e05dbcb1d300204", features = ["std", "_test_utils"] }
+lightning = { path = "../rust-lightning/lightning", features = ["std", "_test_utils"] }
 proptest = "1.0.0"
 regex = "1.5.6"
 
diff --git a/src/builder.rs b/src/builder.rs
@@ -1557,6 +1557,7 @@ fn build_with_store_internal(
 				Arc::clone(&channel_manager),
 				Arc::clone(&keys_manager),
 				Arc::clone(&chain_source),
+				Arc::clone(&tx_broadcaster),
 				Arc::clone(&kv_store),
 				Arc::clone(&config),
 				Arc::clone(&logger),
diff --git a/src/data_store.rs b/src/data_store.rs
@@ -103,7 +103,6 @@ where
 				&self.primary_namespace,
 				&self.secondary_namespace,
 				&store_key,
-				false,
 			)
 			.map_err(|e| {
 				log_error!(
diff --git a/src/event.rs b/src/event.rs
@@ -1289,7 +1289,16 @@ where
 				}
 
 				if let Some(liquidity_source) = self.liquidity_source.as_ref() {
-					liquidity_source.handle_payment_forwarded(next_channel_id).await;
+					if let Some(skimmed_fee_msat) = skimmed_fee_msat {
+						liquidity_source
+							.handle_payment_forwarded(next_channel_id, skimmed_fee_msat)
+							.await;
+					} else {
+						debug_assert!(
+							false,
+							"We expect skimmed_fee_msat to be set since LDK 0.0.122"
+						);
+					}
 				}
 
 				let event = Event::PaymentForwarded {
diff --git a/src/io/sqlite_store/mod.rs b/src/io/sqlite_store/mod.rs
@@ -137,7 +137,7 @@ impl KVStore for SqliteStore {
 	}
 
 	fn remove(
-		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
 	) -> Pin<Box<dyn Future<Output = Result<(), io::Error>> + Send>> {
 		let locking_key = self.build_locking_key(primary_namespace, secondary_namespace, key);
 		let (inner_lock_ref, version) = self.get_new_version_and_lock_ref(locking_key.clone());
@@ -153,7 +153,6 @@ impl KVStore for SqliteStore {
 				&primary_namespace,
 				&secondary_namespace,
 				&key,
-				lazy,
 			)
 		});
 		Box::pin(async move {
@@ -206,7 +205,7 @@ impl KVStoreSync for SqliteStore {
 	}
 
 	fn remove(
-		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
 	) -> io::Result<()> {
 		let locking_key = self.build_locking_key(primary_namespace, secondary_namespace, key);
 		let (inner_lock_ref, version) = self.get_new_version_and_lock_ref(locking_key.clone());
@@ -217,7 +216,6 @@ impl KVStoreSync for SqliteStore {
 			primary_namespace,
 			secondary_namespace,
 			key,
-			lazy,
 		)
 	}
 
@@ -402,7 +400,7 @@ impl SqliteStoreInner {
 
 	fn remove_internal(
 		&self, inner_lock_ref: Arc<Mutex<u64>>, locking_key: String, version: u64,
-		primary_namespace: &str, secondary_namespace: &str, key: &str, _lazy: bool,
+		primary_namespace: &str, secondary_namespace: &str, key: &str,
 	) -> io::Result<()> {
 		check_namespace_key_validity(primary_namespace, secondary_namespace, Some(key), "remove")?;
 
diff --git a/src/io/test_utils.rs b/src/io/test_utils.rs
@@ -55,7 +55,7 @@ pub(crate) fn do_read_write_remove_list_persist<K: KVStoreSync + RefUnwindSafe>(
 	let read_data = kv_store.read(primary_namespace, secondary_namespace, key).unwrap();
 	assert_eq!(data, &*read_data);
 
-	kv_store.remove(primary_namespace, secondary_namespace, key, false).unwrap();
+	kv_store.remove(primary_namespace, secondary_namespace, key).unwrap();
 
 	let listed_keys = kv_store.list(primary_namespace, secondary_namespace).unwrap();
 	assert_eq!(listed_keys.len(), 0);
@@ -71,7 +71,7 @@ pub(crate) fn do_read_write_remove_list_persist<K: KVStoreSync + RefUnwindSafe>(
 	let read_data = kv_store.read(&max_chars, &max_chars, &max_chars).unwrap();
 	assert_eq!(data, &*read_data);
 
-	kv_store.remove(&max_chars, &max_chars, &max_chars, false).unwrap();
+	kv_store.remove(&max_chars, &max_chars, &max_chars).unwrap();
 
 	let listed_keys = kv_store.list(&max_chars, &max_chars).unwrap();
 	assert_eq!(listed_keys.len(), 0);
diff --git a/src/io/utils.rs b/src/io/utils.rs
@@ -32,7 +32,6 @@ use lightning::util::persist::{
 	SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
 };
 use lightning::util::ser::{Readable, ReadableArgs, Writeable};
-use lightning::util::sweep::OutputSweeper;
 use lightning_types::string::PrintableString;
 use rand::{thread_rng, RngCore};
 
@@ -256,10 +255,11 @@ pub(crate) fn read_output_sweeper(
 		kv_store,
 		logger.clone(),
 	);
-	OutputSweeper::read(&mut reader, args).map_err(|e| {
+	let (_, sweeper) = <(_, Sweeper)>::read(&mut reader, args).map_err(|e| {
 		log_error!(logger, "Failed to deserialize OutputSweeper: {}", e);
 		std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize OutputSweeper")
-	})
+	})?;
+	Ok(sweeper)
 }
 
 pub(crate) fn read_node_metrics<L: Deref>(
diff --git a/src/io/vss_store.rs b/src/io/vss_store.rs
@@ -116,7 +116,7 @@ impl KVStoreSync for VssStore {
 	}
 
 	fn remove(
-		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
 	) -> io::Result<()> {
 		let locking_key = self.build_locking_key(primary_namespace, secondary_namespace, key);
 		let (inner_lock_ref, version) = self.get_new_version_and_lock_ref(locking_key.clone());
@@ -127,7 +127,6 @@ impl KVStoreSync for VssStore {
 			primary_namespace,
 			secondary_namespace,
 			key,
-			lazy,
 		);
 		self.runtime.block_on(fut)
 	}
@@ -174,7 +173,7 @@ impl KVStore for VssStore {
 		})
 	}
 	fn remove(
-		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
 	) -> Pin<Box<dyn Future<Output = Result<(), io::Error>> + Send>> {
 		let locking_key = self.build_locking_key(primary_namespace, secondary_namespace, key);
 		let (inner_lock_ref, version) = self.get_new_version_and_lock_ref(locking_key.clone());
@@ -191,7 +190,6 @@ impl KVStore for VssStore {
 					&primary_namespace,
 					&secondary_namespace,
 					&key,
-					lazy,
 				)
 				.await
 		})
@@ -369,7 +367,7 @@ impl VssStoreInner {
 
 	async fn remove_internal(
 		&self, inner_lock_ref: Arc<tokio::sync::Mutex<u64>>, locking_key: String, version: u64,
-		primary_namespace: &str, secondary_namespace: &str, key: &str, _lazy: bool,
+		primary_namespace: &str, secondary_namespace: &str, key: &str,
 	) -> io::Result<()> {
 		check_namespace_key_validity(primary_namespace, secondary_namespace, Some(key), "remove")?;
 
diff --git a/src/liquidity.rs b/src/liquidity.rs
@@ -43,7 +43,9 @@ use crate::chain::ChainSource;
 use crate::connection::ConnectionManager;
 use crate::logger::{log_debug, log_error, log_info, LdkLogger, Logger};
 use crate::runtime::Runtime;
-use crate::types::{ChannelManager, DynStore, KeysManager, LiquidityManager, PeerManager, Wallet};
+use crate::types::{
+	Broadcaster, ChannelManager, DynStore, KeysManager, LiquidityManager, PeerManager, Wallet,
+};
 use crate::{total_anchor_channels_reserve_sats, Config, Error};
 
 const LIQUIDITY_REQUEST_TIMEOUT_SECS: u64 = 5;
@@ -141,6 +143,7 @@ where
 	channel_manager: Arc<ChannelManager>,
 	keys_manager: Arc<KeysManager>,
 	chain_source: Arc<ChainSource>,
+	tx_broadcaster: Arc<Broadcaster>,
 	kv_store: Arc<DynStore>,
 	config: Arc<Config>,
 	logger: L,
@@ -152,7 +155,8 @@ where
 {
 	pub(crate) fn new(
 		wallet: Arc<Wallet>, channel_manager: Arc<ChannelManager>, keys_manager: Arc<KeysManager>,
-		chain_source: Arc<ChainSource>, kv_store: Arc<DynStore>, config: Arc<Config>, logger: L,
+		chain_source: Arc<ChainSource>, tx_broadcaster: Arc<Broadcaster>, kv_store: Arc<DynStore>,
+		config: Arc<Config>, logger: L,
 	) -> Self {
 		let lsps1_client = None;
 		let lsps2_client = None;
@@ -165,6 +169,7 @@ where
 			channel_manager,
 			keys_manager,
 			chain_source,
+			tx_broadcaster,
 			kv_store,
 			config,
 			logger,
@@ -241,6 +246,7 @@ where
 				Some(Arc::clone(&self.chain_source)),
 				None,
 				Arc::clone(&self.kv_store),
+				Arc::clone(&self.tx_broadcaster),
 				liquidity_service_config,
 				liquidity_client_config,
 			)
@@ -1302,10 +1308,14 @@ where
 		}
 	}
 
-	pub(crate) async fn handle_payment_forwarded(&self, next_channel_id: Option<ChannelId>) {
+	pub(crate) async fn handle_payment_forwarded(
+		&self, next_channel_id: Option<ChannelId>, skimmed_fee_msat: u64,
+	) {
 		if let Some(next_channel_id) = next_channel_id {
 			if let Some(lsps2_service_handler) = self.liquidity_manager.lsps2_service_handler() {
-				if let Err(e) = lsps2_service_handler.payment_forwarded(next_channel_id).await {
+				if let Err(e) =
+					lsps2_service_handler.payment_forwarded(next_channel_id, skimmed_fee_msat).await
+				{
 					log_error!(
 						self.logger,
 						"LSPS2 service failed to handle PaymentForwarded: {:?}",
diff --git a/src/types.rs b/src/types.rs
@@ -76,7 +76,8 @@ pub(crate) type LiquidityManager = lightning_liquidity::LiquidityManager<
 	Arc<ChannelManager>,
 	Arc<ChainSource>,
 	Arc<DynStore>,
-	Arc<DefaultTimeProvider>,
+	DefaultTimeProvider,
+	Arc<Broadcaster>,
 >;
 
 pub(crate) type ChannelManager = lightning::ln::channelmanager::ChannelManager<
diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs
@@ -916,7 +916,7 @@ impl SignerProvider for WalletKeysManager {
 
 impl ChangeDestinationSource for WalletKeysManager {
 	fn get_change_destination_script<'a>(
-		&self,
+		&'a self,
 	) -> Pin<Box<dyn Future<Output = Result<ScriptBuf, ()>> + Send + 'a>> {
 		let wallet = Arc::clone(&self.wallet);
 		let logger = Arc::clone(&self.logger);
diff --git a/tests/common/mod.rs b/tests/common/mod.rs
@@ -1240,14 +1240,14 @@ impl KVStore for TestSyncStore {
 		})
 	}
 	fn remove(
-		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
 	) -> Pin<Box<dyn Future<Output = Result<(), io::Error>> + Send>> {
 		let primary_namespace = primary_namespace.to_string();
 		let secondary_namespace = secondary_namespace.to_string();
 		let key = key.to_string();
 		let inner = Arc::clone(&self.inner);
 		let fut = tokio::task::spawn_blocking(move || {
-			inner.remove_internal(&primary_namespace, &secondary_namespace, &key, lazy)
+			inner.remove_internal(&primary_namespace, &secondary_namespace, &key)
 		});
 		Box::pin(async move {
 			fut.await.unwrap_or_else(|e| {
@@ -1288,9 +1288,9 @@ impl KVStoreSync for TestSyncStore {
 	}
 
 	fn remove(
-		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
 	) -> lightning::io::Result<()> {
-		self.inner.remove_internal(primary_namespace, secondary_namespace, key, lazy)
+		self.inner.remove_internal(primary_namespace, secondary_namespace, key)
 	}
 
 	fn list(
@@ -1428,25 +1428,15 @@ impl TestSyncStoreInner {
 	}
 
 	fn remove_internal(
-		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
 	) -> lightning::io::Result<()> {
 		let _guard = self.serializer.write().unwrap();
 		let fs_res =
-			KVStoreSync::remove(&self.fs_store, primary_namespace, secondary_namespace, key, lazy);
-		let sqlite_res = KVStoreSync::remove(
-			&self.sqlite_store,
-			primary_namespace,
-			secondary_namespace,
-			key,
-			lazy,
-		);
-		let test_res = KVStoreSync::remove(
-			&self.test_store,
-			primary_namespace,
-			secondary_namespace,
-			key,
-			lazy,
-		);
+			KVStoreSync::remove(&self.fs_store, primary_namespace, secondary_namespace, key);
+		let sqlite_res =
+			KVStoreSync::remove(&self.sqlite_store, primary_namespace, secondary_namespace, key);
+		let test_res =
+			KVStoreSync::remove(&self.test_store, primary_namespace, secondary_namespace, key);
 
 		assert!(!self
 			.do_list(primary_namespace, secondary_namespace)

Original file line number	Diff line number	Diff line change
`@@ -103,7 +103,6 @@ where`
`103`	`103`	`&self.primary_namespace,`
`104`	`104`	`&self.secondary_namespace,`
`105`	`105`	`&store_key,`
`106`		`- false,`
`107`	`106`	`)`
`108`	`107`	`.map_err(\|e\| {`
`109`	`108`	`log_error!(`