Merge pull request #782 from joostjager/chain-mon-deferred-writes

Enable deferred writing

Author: tnull

Cargo.toml

@@ -39,17 +39,17 @@ default = []
 #lightning-liquidity = { version = "0.2.0", features = ["std"] }
 #lightning-macros = { version = "0.2.0" }
 
-lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "49912057895ddfbd69d503de67c80d5576c09953", features = ["std"] }
-lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "49912057895ddfbd69d503de67c80d5576c09953" }
-lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "49912057895ddfbd69d503de67c80d5576c09953", features = ["std"] }
-lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "49912057895ddfbd69d503de67c80d5576c09953" }
-lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "49912057895ddfbd69d503de67c80d5576c09953", features = ["tokio"] }
-lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "49912057895ddfbd69d503de67c80d5576c09953" }
-lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "49912057895ddfbd69d503de67c80d5576c09953" }
-lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "49912057895ddfbd69d503de67c80d5576c09953", features = ["rest-client", "rpc-client", "tokio"] }
-lightning-transaction-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "49912057895ddfbd69d503de67c80d5576c09953", features = ["esplora-async-https", "time", "electrum-rustls-ring"] }
-lightning-liquidity = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "49912057895ddfbd69d503de67c80d5576c09953", features = ["std"] }
-lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "49912057895ddfbd69d503de67c80d5576c09953" }
+lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "dcf0c203e166da2348bef12b2e5eff4a250cdec7", features = ["std"] }
+lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "dcf0c203e166da2348bef12b2e5eff4a250cdec7" }
+lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "dcf0c203e166da2348bef12b2e5eff4a250cdec7", features = ["std"] }
+lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "dcf0c203e166da2348bef12b2e5eff4a250cdec7" }
+lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "dcf0c203e166da2348bef12b2e5eff4a250cdec7", features = ["tokio"] }
+lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "dcf0c203e166da2348bef12b2e5eff4a250cdec7" }
+lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "dcf0c203e166da2348bef12b2e5eff4a250cdec7" }
+lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "dcf0c203e166da2348bef12b2e5eff4a250cdec7", features = ["rest-client", "rpc-client", "tokio"] }
+lightning-transaction-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "dcf0c203e166da2348bef12b2e5eff4a250cdec7", features = ["esplora-async-https", "time", "electrum-rustls-ring"] }
+lightning-liquidity = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "dcf0c203e166da2348bef12b2e5eff4a250cdec7", features = ["std"] }
+lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "dcf0c203e166da2348bef12b2e5eff4a250cdec7" }
 
 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"]}
@@ -79,13 +79,13 @@ async-trait = { version = "0.1", default-features = false }
 vss-client = { package = "vss-client-ng", version = "0.5" }
 prost = { version = "0.11.6", default-features = false}
 #bitcoin-payment-instructions = { version = "0.6" }
-bitcoin-payment-instructions = { git = "https://github.com/tnull/bitcoin-payment-instructions", rev = "e9d7c07d7affc7714b023c853a65771e45277467" }
+bitcoin-payment-instructions = { git = "https://github.com/joostjager/bitcoin-payment-instructions", branch = "ldk-dcf0c203e166da2348bef12b2e5eff4a250cdec7" }
 
 [target.'cfg(windows)'.dependencies]
 winapi = { version = "0.3", features = ["winbase"] }
 
 [dev-dependencies]
-lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "49912057895ddfbd69d503de67c80d5576c09953", features = ["std", "_test_utils"] }
+lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "dcf0c203e166da2348bef12b2e5eff4a250cdec7", features = ["std", "_test_utils"] }
 rand = { version = "0.9.2", default-features = false, features = ["std", "thread_rng", "os_rng"] }
 proptest = "1.0.0"
 regex = "1.5.6"

src/builder.rs

@@ -76,9 +76,9 @@ use crate::peer_store::PeerStore;
 use crate::runtime::{Runtime, RuntimeSpawner};
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
-	AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreWrapper, GossipSync, Graph,
-	KeysManager, MessageRouter, OnionMessenger, PaymentStore, PeerManager, PendingPaymentStore,
-	Persister, SyncAndAsyncKVStore,
+	AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreRef, DynStoreWrapper,
+	GossipSync, Graph, KeysManager, MessageRouter, OnionMessenger, PaymentStore, PeerManager,
+	PendingPaymentStore, SyncAndAsyncKVStore,
 };
 use crate::wallet::persist::KVStoreWalletPersister;
 use crate::wallet::Wallet;
@@ -1495,7 +1495,7 @@ fn build_with_store_internal(
 
 	let peer_storage_key = keys_manager.get_peer_storage_key();
 	let monitor_reader = Arc::new(AsyncPersister::new(
-		Arc::clone(&kv_store),
+		DynStoreRef(Arc::clone(&kv_store)),
 		RuntimeSpawner::new(Arc::clone(&runtime)),
 		Arc::clone(&logger),
 		PERSISTER_MAX_PENDING_UPDATES,
@@ -1508,7 +1508,7 @@ fn build_with_store_internal(
 	// Read ChannelMonitors and the NetworkGraph
 	let kv_store_ref = Arc::clone(&kv_store);
 	let logger_ref = Arc::clone(&logger);
-	let (monitor_read_res, network_graph_res) = runtime.block_on(async move {
+	let (monitor_read_res, network_graph_res) = runtime.block_on(async {
 		tokio::join!(
 			monitor_reader.read_all_channel_monitors_with_updates_parallel(),
 			read_network_graph(&*kv_store_ref, logger_ref),
@@ -1528,26 +1528,21 @@ fn build_with_store_internal(
 		},
 	};
 
-	let persister = Arc::new(Persister::new(
-		Arc::clone(&kv_store),
-		Arc::clone(&logger),
-		PERSISTER_MAX_PENDING_UPDATES,
-		Arc::clone(&keys_manager),
-		Arc::clone(&keys_manager),
-		Arc::clone(&tx_broadcaster),
-		Arc::clone(&fee_estimator),
-	));
-
 	// Initialize the ChainMonitor
-	let chain_monitor: Arc<ChainMonitor> = Arc::new(chainmonitor::ChainMonitor::new(
-		Some(Arc::clone(&chain_source)),
-		Arc::clone(&tx_broadcaster),
-		Arc::clone(&logger),
-		Arc::clone(&fee_estimator),
-		Arc::clone(&persister),
-		Arc::clone(&keys_manager),
-		peer_storage_key,
-	));
+	let chain_monitor: Arc<ChainMonitor> = {
+		let persister = Arc::try_unwrap(monitor_reader)
+			.unwrap_or_else(|_| panic!("Arc<AsyncPersister> should have no other references"));
+		Arc::new(chainmonitor::ChainMonitor::new_async_beta(
+			Some(Arc::clone(&chain_source)),
+			Arc::clone(&tx_broadcaster),
+			Arc::clone(&logger),
+			Arc::clone(&fee_estimator),
+			persister,
+			Arc::clone(&keys_manager),
+			peer_storage_key,
+			true,
+		))
+	};
 
 	// Initialize the network graph, scorer, and router
 	let network_graph = match network_graph_res {

src/event.rs

@@ -691,6 +691,26 @@ where
 					if info.status == PaymentStatus::Succeeded
 						|| matches!(info.kind, PaymentKind::Spontaneous { .. })
 					{
+						let stored_preimage = match info.kind {
+							PaymentKind::Bolt11 { preimage, .. }
+							| PaymentKind::Bolt11Jit { preimage, .. }
+							| PaymentKind::Bolt12Offer { preimage, .. }
+							| PaymentKind::Bolt12Refund { preimage, .. }
+							| PaymentKind::Spontaneous { preimage, .. } => preimage,
+							_ => None,
+						};
+
+						if let Some(preimage) = stored_preimage {
+							log_info!(
+								self.logger,
+								"Re-claiming previously succeeded payment with hash {} of {}msat",
+								hex_utils::to_string(&payment_hash.0),
+								amount_msat,
+							);
+							self.channel_manager.claim_funds(preimage);
+							return Ok(());
+						}
+
 						log_info!(
 							self.logger,
 							"Refused duplicate inbound payment from payment hash {} of {}msat",

src/types.rs

@@ -23,9 +23,7 @@ use lightning::routing::gossip;
 use lightning::routing::router::DefaultRouter;
 use lightning::routing::scoring::{CombinedScorer, ProbabilisticScoringFeeParameters};
 use lightning::sign::InMemorySigner;
-use lightning::util::persist::{
-	KVStore, KVStoreSync, MonitorUpdatingPersister, MonitorUpdatingPersisterAsync,
-};
+use lightning::util::persist::{KVStore, KVStoreSync, MonitorUpdatingPersisterAsync};
 use lightning::util::ser::{Readable, Writeable, Writer};
 use lightning::util::sweep::OutputSweeper;
 use lightning_block_sync::gossip::GossipVerifier;
@@ -135,6 +133,39 @@ impl<'a> KVStoreSync for dyn DynStoreTrait + 'a {
 
 pub(crate) type DynStore = dyn DynStoreTrait;
 
+// Newtype wrapper that implements `KVStore` for `Arc<DynStore>`. This is needed because `KVStore`
+// methods return `impl Future`, which is not object-safe. `DynStoreTrait` works around this by
+// returning `Pin<Box<dyn Future>>` instead, and this wrapper bridges the two by delegating
+// `KVStore` methods to the corresponding `DynStoreTrait::*_async` methods.
+#[derive(Clone)]
+pub(crate) struct DynStoreRef(pub(crate) Arc<DynStore>);
+
+impl KVStore for DynStoreRef {
+	fn read(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+	) -> impl Future<Output = Result<Vec<u8>, bitcoin::io::Error>> + Send + 'static {
+		DynStoreTrait::read_async(&*self.0, primary_namespace, secondary_namespace, key)
+	}
+
+	fn write(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
+	) -> impl Future<Output = Result<(), bitcoin::io::Error>> + Send + 'static {
+		DynStoreTrait::write_async(&*self.0, primary_namespace, secondary_namespace, key, buf)
+	}
+
+	fn remove(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+	) -> impl Future<Output = Result<(), bitcoin::io::Error>> + Send + 'static {
+		DynStoreTrait::remove_async(&*self.0, primary_namespace, secondary_namespace, key, lazy)
+	}
+
+	fn list(
+		&self, primary_namespace: &str, secondary_namespace: &str,
+	) -> impl Future<Output = Result<Vec<String>, bitcoin::io::Error>> + Send + 'static {
+		DynStoreTrait::list_async(&*self.0, primary_namespace, secondary_namespace)
+	}
+}
+
 pub(crate) struct DynStoreWrapper<T: SyncAndAsyncKVStore + Send + Sync>(pub(crate) T);
 
 impl<T: SyncAndAsyncKVStore + Send + Sync> DynStoreTrait for DynStoreWrapper<T> {
@@ -188,7 +219,7 @@ impl<T: SyncAndAsyncKVStore + Send + Sync> DynStoreTrait for DynStoreWrapper<T>
 }
 
 pub(crate) type AsyncPersister = MonitorUpdatingPersisterAsync<
-	Arc<DynStore>,
+	DynStoreRef,
 	RuntimeSpawner,
 	Arc<Logger>,
 	Arc<KeysManager>,
@@ -197,22 +228,21 @@ pub(crate) type AsyncPersister = MonitorUpdatingPersisterAsync<
 	Arc<OnchainFeeEstimator>,
 >;
 
-pub type Persister = MonitorUpdatingPersister<
-	Arc<DynStore>,
-	Arc<Logger>,
-	Arc<KeysManager>,
-	Arc<KeysManager>,
-	Arc<Broadcaster>,
-	Arc<OnchainFeeEstimator>,
->;
-
 pub(crate) type ChainMonitor = chainmonitor::ChainMonitor<
 	InMemorySigner,
 	Arc<ChainSource>,
 	Arc<Broadcaster>,
 	Arc<OnchainFeeEstimator>,
 	Arc<Logger>,
-	Arc<Persister>,
+	chainmonitor::AsyncPersister<
+		DynStoreRef,
+		RuntimeSpawner,
+		Arc<Logger>,
+		Arc<KeysManager>,
+		Arc<KeysManager>,
+		Arc<Broadcaster>,
+		Arc<OnchainFeeEstimator>,
+	>,
 	Arc<KeysManager>,
 >;
 

tests/common/mod.rs

@@ -1262,8 +1262,8 @@ pub(crate) async fn do_channel_full_cycle<E: ElectrumApi>(
 	);
 
 	println!("\nB close_channel (force: {})", force_close);
+	tokio::time::sleep(Duration::from_secs(1)).await;
 	if force_close {
-		tokio::time::sleep(Duration::from_secs(1)).await;
 		node_a.force_close_channel(&user_channel_id_a, node_b.node_id(), None).unwrap();
 	} else {
 		node_a.close_channel(&user_channel_id_a, node_b.node_id()).unwrap();