diff --git a/src/builder.rs b/src/builder.rs
index 21f15508b4..da03039d34 100644
--- a/src/builder.rs
+++ b/src/builder.rs
@@ -105,7 +105,7 @@ use crate::probing::{
 	HighDegreeStrategy, Prober, ProbingConfig, ProbingStrategy, ProbingStrategyKind,
 	RandomWalkStrategy,
 };
-use crate::runtime::{Runtime, RuntimeSpawner};
+use crate::runtime::{MonitorPersistenceSpawner, Runtime};
 use crate::tx_broadcaster::TransactionBroadcaster;
 use crate::types::{
 	AsyncPersister, ChainMonitor, ChannelManager, ChannelTxFactsStore, DynStore, DynStoreRef,
@@ -2017,7 +2017,7 @@ fn build_with_store_internal(
 	let peer_storage_key = keys_manager.get_peer_storage_key();
 	let monitor_reader = Arc::new(AsyncPersister::new(
 		DynStoreRef(Arc::clone(&kv_store)),
-		RuntimeSpawner::new(Arc::clone(&runtime)),
+		MonitorPersistenceSpawner::new(&runtime),
 		Arc::clone(&logger),
 		PERSISTER_MAX_PENDING_UPDATES,
 		Arc::clone(&keys_manager),
diff --git a/src/runtime.rs b/src/runtime.rs
index 8f9e94ccf3..effdae7cde 100644
--- a/src/runtime.rs
+++ b/src/runtime.rs
@@ -26,6 +26,8 @@ pub(crate) struct Runtime {
 	background_tasks: Mutex<JoinSet<()>>,
 	cancellable_background_tasks: Mutex<CancellableBackgroundTasks>,
 	background_processor_task: Mutex<Option<JoinHandle<()>>>,
+	/// Cancels the tasks of every [`MonitorPersistenceSpawner`] once the [`Runtime`] is dropped.
+	monitor_persistence_cancellation_token: CancellationToken,
 	logger: Arc<Logger>,
 }
 
@@ -77,12 +79,14 @@ impl Runtime {
 		let background_tasks = Mutex::new(JoinSet::new());
 		let cancellable_background_tasks = Mutex::new(CancellableBackgroundTasks::new());
 		let background_processor_task = Mutex::new(None);
+		let monitor_persistence_cancellation_token = CancellationToken::new();
 
 		Ok(Self {
 			mode,
 			background_tasks,
 			cancellable_background_tasks,
 			background_processor_task,
+			monitor_persistence_cancellation_token,
 			logger,
 		})
 	}
@@ -92,12 +96,14 @@ impl Runtime {
 		let background_tasks = Mutex::new(JoinSet::new());
 		let cancellable_background_tasks = Mutex::new(CancellableBackgroundTasks::new());
 		let background_processor_task = Mutex::new(None);
+		let monitor_persistence_cancellation_token = CancellationToken::new();
 
 		Self {
 			mode,
 			background_tasks,
 			cancellable_background_tasks,
 			background_processor_task,
+			monitor_persistence_cancellation_token,
 			logger,
 		}
 	}
@@ -295,6 +301,12 @@ impl Runtime {
 	}
 }
 
+impl Drop for Runtime {
+	fn drop(&mut self) {
+		self.monitor_persistence_cancellation_token.cancel();
+	}
+}
+
 enum RuntimeMode {
 	Owned(tokio::runtime::Runtime),
 	Handle(tokio::runtime::Handle),
@@ -384,16 +396,19 @@ impl Drop for StoreRuntime {
 	}
 }
 
+#[cfg(feature = "chain-bitcoind")]
 pub(crate) struct RuntimeSpawner {
 	runtime: Arc<Runtime>,
 }
 
+#[cfg(feature = "chain-bitcoind")]
 impl RuntimeSpawner {
 	pub(crate) fn new(runtime: Arc<Runtime>) -> Self {
 		Self { runtime }
 	}
 }
 
+#[cfg(feature = "chain-bitcoind")]
 impl FutureSpawner for RuntimeSpawner {
 	type E = tokio::sync::oneshot::error::RecvError;
 	type SpawnedFutureResult<O> = tokio::sync::oneshot::Receiver<O>;
@@ -410,6 +425,52 @@ impl FutureSpawner for RuntimeSpawner {
 	}
 }
 
+/// The [`FutureSpawner`] of our `ChannelMonitor` persister.
+///
+/// The persister reports each completed `ChannelMonitor` write to the `ChainMonitor` from a task
+/// spawned here. If that task never ran, the channel would keep waiting on the write until the
+/// node is rebuilt, so unlike [`Runtime::spawn_cancellable_background_task`] its tasks are neither
+/// cancelled nor refused when the node is stopped, and only cancelled once the [`Runtime`] is
+/// dropped.
+pub(crate) struct MonitorPersistenceSpawner {
+	// Only a handle rather than an `Arc<Runtime>`: the spawned tasks own the persister, which owns
+	// this spawner, so they must not keep the runtime alive.
+	handle: tokio::runtime::Handle,
+	cancellation_token: CancellationToken,
+}
+
+impl MonitorPersistenceSpawner {
+	pub(crate) fn new(runtime: &Runtime) -> Self {
+		Self {
+			handle: runtime.handle().clone(),
+			cancellation_token: runtime.monitor_persistence_cancellation_token.clone(),
+		}
+	}
+}
+
+impl FutureSpawner for MonitorPersistenceSpawner {
+	type E = tokio::sync::oneshot::error::RecvError;
+	type SpawnedFutureResult<O> = tokio::sync::oneshot::Receiver<O>;
+	fn spawn<O: Send + 'static, F: Future<Output = O> + Send + 'static>(
+		&self, future: F,
+	) -> Self::SpawnedFutureResult<O> {
+		let (result, output) = tokio::sync::oneshot::channel();
+		let cancellation_token = self.cancellation_token.clone();
+		self.handle.spawn(async move {
+			tokio::select! {
+				biased;
+				_ = cancellation_token.cancelled() => {},
+				// We don't care if the send works or not, if the receiver is dropped its not our
+				// problem.
+				res = future => {
+					let _ = result.send(res);
+				},
+			}
+		});
+		output
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	use tokio::sync::{mpsc, oneshot};
@@ -617,4 +678,83 @@ mod tests {
 		});
 		assert!(task_was_dropped, "dropping the runtime should cancel detached tasks");
 	}
+
+	#[test]
+	fn monitor_persistence_tasks_are_not_cancelled_by_stop() {
+		let runtime = test_runtime();
+		let spawner = MonitorPersistenceSpawner::new(&runtime);
+
+		// A write that is still in flight when the node stops.
+		let (in_flight_started_sender, in_flight_started_receiver) = oneshot::channel();
+		let (in_flight_release_sender, in_flight_release_receiver) = oneshot::channel::<()>();
+		let in_flight = spawner.spawn(async move {
+			let _ = in_flight_started_sender.send(());
+			in_flight_release_receiver.await.expect("test should still be running");
+		});
+		runtime.block_on(async {
+			in_flight_started_receiver.await.expect("in-flight task should start");
+		});
+
+		runtime.abort_cancellable_background_tasks();
+
+		// A write spawned while the node is stopping, e.g., by the background processor's final
+		// flush.
+		let late = spawner.spawn(async {});
+		let _ = in_flight_release_sender.send(());
+		let (in_flight_res, late_res) = runtime.block_on(async {
+			tokio::time::timeout(Duration::from_secs(1), async { tokio::join!(in_flight, late) })
+				.await
+				.expect("monitor persistence tasks should complete after stop")
+		});
+		assert!(in_flight_res.is_ok(), "in-flight task should complete after stop");
+		assert!(late_res.is_ok(), "task spawned during stop should complete");
+	}
+
+	#[test]
+	fn dropping_runtime_cancels_monitor_persistence_tasks() {
+		fn spawn_pending_task_owning_spawner(
+			runtime: &Runtime,
+		) -> (oneshot::Receiver<()>, oneshot::Receiver<()>) {
+			// Like the persister's tasks, the task owns the spawner that spawned it.
+			let spawner = Arc::new(MonitorPersistenceSpawner::new(runtime));
+			let task_spawner = Arc::clone(&spawner);
+			let (started_sender, started_receiver) = oneshot::channel();
+			let (dropped_sender, dropped_receiver) = oneshot::channel();
+			let _ = spawner.spawn(async move {
+				let _spawner = task_spawner;
+				let _drop_notifier = DropNotifier(Some(dropped_sender));
+				let _ = started_sender.send(());
+				std::future::pending::<()>().await;
+			});
+			(started_receiver, dropped_receiver)
+		}
+
+		// A runtime we own.
+		let runtime = test_runtime();
+		let (started_receiver, dropped_receiver) = spawn_pending_task_owning_spawner(&runtime);
+		started_receiver.blocking_recv().expect("task should start");
+		drop(runtime);
+		assert!(
+			dropped_receiver.blocking_recv().is_ok(),
+			"task should be dropped with the runtime"
+		);
+
+		// An externally supplied runtime handle.
+		let tokio_runtime =
+			tokio::runtime::Builder::new_multi_thread().enable_all().build().expect("test runtime");
+		let runtime = Runtime::with_handle(
+			tokio_runtime.handle().clone(),
+			Arc::new(Logger::new_log_facade()),
+		);
+		let (started_receiver, dropped_receiver) = spawn_pending_task_owning_spawner(&runtime);
+		started_receiver.blocking_recv().expect("task should start");
+		drop(runtime);
+		let task_was_dropped = tokio_runtime.block_on(async {
+			matches!(
+				tokio::time::timeout(Duration::from_secs(1), dropped_receiver).await,
+				Ok(Ok(()))
+			)
+		});
+		assert!(task_was_dropped, "dropping the runtime should cancel monitor persistence tasks");
+	}
 }
diff --git a/src/types.rs b/src/types.rs
index 9eb3fdaa8c..0777943e8c 100644
--- a/src/types.rs
+++ b/src/types.rs
@@ -46,7 +46,7 @@ use crate::message_handler::NodeCustomMessageHandler;
 use crate::payment::{
 	ChannelPairForwardingStats, ForwardedPaymentDetails, PaymentDetails, PendingPaymentDetails,
 };
-use crate::runtime::RuntimeSpawner;
+use crate::runtime::MonitorPersistenceSpawner;
 use crate::wallet::provenance::ChannelTxFacts;
 
 #[cfg(feature = "uniffi")]
@@ -210,7 +210,7 @@ impl<T: PaginatedKVStore + Send + Sync> DynStoreTrait for DynStoreWrapper<T> {
 
 pub(crate) type AsyncPersister = MonitorUpdatingPersisterAsync<
 	DynStoreRef,
-	RuntimeSpawner,
+	MonitorPersistenceSpawner,
 	Arc<Logger>,
 	Arc<KeysManager>,
 	Arc<KeysManager>,
@@ -226,7 +226,7 @@ pub(crate) type ChainMonitor = chainmonitor::ChainMonitor<
 	Arc<Logger>,
 	chainmonitor::AsyncPersister<
 		DynStoreRef,
-		RuntimeSpawner,
+		MonitorPersistenceSpawner,
 		Arc<Logger>,
 		Arc<KeysManager>,
 		Arc<KeysManager>,
diff --git a/tests/integration_tests_monitor_persistence.rs b/tests/integration_tests_monitor_persistence.rs
new file mode 100755
index 0000000000..017d8d9065
--- /dev/null
+++ b/tests/integration_tests_monitor_persistence.rs
@@ -0,0 +1,342 @@
+// This file is Copyright its original authors, visible in version control history.
+//
+// This file is licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license <LICENSE-MIT or
+// http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
+// accordance with one or both of these licenses.
+
+//! Tests that `ChannelMonitor` persistence which is in flight, or which is started, while
+//! [`Node::stop`] runs still completes, so the channel makes progress once the same `Node` is
+//! started again.
+//!
+//! [`Node::stop`]: ldk_node::Node::stop
+
+mod common;
+
+use std::future::Future;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use bitcoin::Amount;
+use common::logging::{CollectingLogWriter, TestLogWriter};
+use common::{
+	expect_channel_ready_event, generate_blocks_and_wait, open_channel, open_channel_push_amt,
+	premine_and_distribute_funds, random_config, setup_bitcoind_and_electrsd, setup_node,
+	setup_node_with_store, wait_for_channel_ready_to_send, TestChainSource, TestNode,
+};
+use ldk_node::io::sqlite_store::SqliteStore;
+use ldk_node::Event;
+use lightning::ln::channelmanager::PaymentId;
+use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore, PaginatedListResponse};
+use lightning_invoice::{Bolt11InvoiceDescription, Description};
+
+const PAYMENT_WAIT: Duration = Duration::from_secs(30);
+
+#[derive(Clone, Copy, PartialEq)]
+enum HeldWrite {
+	/// The next `ChannelMonitorUpdate` write, held after it was committed.
+	MonitorUpdateCommitted,
+	/// The next `ChannelMonitorUpdate` write, held before it is handed to the store, as a store
+	/// that only issues writes once their future is polled would leave it.
+	MonitorUpdateUnissued,
+	/// The next full `ChannelMonitor` write, held after it was committed.
+	MonitorCommitted,
+	/// The next `ChannelManager` write, held after it was committed.
+	ChannelManagerCommitted,
+}
+
+/// A [`SqliteStore`] that can hold one write until it is released.
+#[derive(Clone)]
+struct HoldingStore {
+	inner: Arc<SqliteStore>,
+	armed: Arc<Mutex<Option<HeldWrite>>>,
+	reached: Arc<tokio::sync::Notify>,
+	release: Arc<tokio::sync::Semaphore>,
+}
+
+impl HoldingStore {
+	fn new(storage_dir_path: &str) -> Self {
+		let inner = Arc::new(
+			SqliteStore::new(
+				storage_dir_path.into(),
+				Some("holding_store.sqlite".to_string()),
+				Some("holding_kv".to_string()),
+			)
+			.unwrap(),
+		);
+		Self {
+			inner,
+			armed: Arc::new(Mutex::new(None)),
+			reached: Arc::new(tokio::sync::Notify::new()),
+			release: Arc::new(tokio::sync::Semaphore::new(0)),
+		}
+	}
+
+	fn hold(&self, write: HeldWrite) {
+		*self.armed.lock().unwrap() = Some(write);
+	}
+
+	async fn wait_until_held(&self) {
+		tokio::time::timeout(Duration::from_secs(60), self.reached.notified())
+			.await
+			.expect("the held write was reached");
+	}
+
+	fn release(&self) {
+		self.release.add_permits(1);
+	}
+}
+
+impl KVStore for HoldingStore {
+	fn read(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
+	) -> impl Future<Output = Result<Vec<u8>, lightning::io::Error>> + 'static + Send {
+		KVStore::read(&*self.inner, primary_namespace, secondary_namespace, key)
+	}
+
+	fn write(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
+	) -> impl Future<Output = Result<(), lightning::io::Error>> + 'static + Send {
+		let held = {
+			let mut armed = self.armed.lock().unwrap();
+			let hit = match *armed {
+				Some(HeldWrite::MonitorUpdateCommitted)
+				| Some(HeldWrite::MonitorUpdateUnissued) => primary_namespace == "monitor_updates",
+				Some(HeldWrite::MonitorCommitted) => primary_namespace == "monitors",
+				Some(HeldWrite::ChannelManagerCommitted) => {
+					primary_namespace.is_empty()
+						&& secondary_namespace.is_empty()
+						&& key == "manager"
+				},
+				None => false,
+			};
+			if hit {
+				armed.take()
+			} else {
+				None
+			}
+		};
+		// Like `SqliteStore` itself, issue the write when asked unless we model a store that
+		// only issues it once polled.
+		let (issued, unissued) = if held == Some(HeldWrite::MonitorUpdateUnissued) {
+			(None, Some(buf))
+		} else {
+			let issued =
+				KVStore::write(&*self.inner, primary_namespace, secondary_namespace, key, buf);
+			(Some(issued), None)
+		};
+		let inner = Arc::clone(&self.inner);
+		let reached = Arc::clone(&self.reached);
+		let release = Arc::clone(&self.release);
+		let (primary_namespace, secondary_namespace, key) =
+			(primary_namespace.to_string(), secondary_namespace.to_string(), key.to_string());
+		async move {
+			if let Some(issued) = issued {
+				issued.await?;
+			}
+			if held.is_some() {
+				reached.notify_one();
+				release.acquire().await.expect("semaphore open").forget();
+			}
+			match unissued {
+				Some(buf) => {
+					KVStore::write(&*inner, &primary_namespace, &secondary_namespace, &key, buf)
+						.await
+				},
+				None => Ok(()),
+			}
+		}
+	}
+
+	fn remove(
+		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
+	) -> impl Future<Output = Result<(), lightning::io::Error>> + 'static + Send {
+		KVStore::remove(&*self.inner, primary_namespace, secondary_namespace, key, lazy)
+	}
+
+	fn list(
+		&self, primary_namespace: &str, secondary_namespace: &str,
+	) -> impl Future<Output = Result<Vec<String>, lightning::io::Error>> + 'static + Send {
+		KVStore::list(&*self.inner, primary_namespace, secondary_namespace)
+	}
+}
+
+impl PaginatedKVStore for HoldingStore {
+	fn list_paginated(
+		&self, primary_namespace: &str, secondary_namespace: &str, page_token: Option<PageToken>,
+	) -> impl Future<Output = Result<PaginatedListResponse, lightning::io::Error>> + 'static + Send
+	{
+		PaginatedKVStore::list_paginated(
+			&*self.inner,
+			primary_namespace,
+			secondary_namespace,
+			page_token,
+		)
+	}
+}
+
+/// Has `payer` pay a new invoice of `payee`.
+fn send_payment(payer: &TestNode, payee: &TestNode, amount_msat: u64) -> PaymentId {
+	let description =
+		Bolt11InvoiceDescription::Direct(Description::new("monitor persistence".into()).unwrap());
+	let invoice = payee.bolt11_payment().receive(amount_msat, &description.into(), 3600).unwrap();
+	payer.bolt11_payment().send(&invoice, None).unwrap()
+}
+
+/// Handles events on `node` until `payment_id` succeeds or fails, or `PAYMENT_WAIT` passes.
+async fn payment_succeeds(node: &TestNode, payment_id: PaymentId) -> bool {
+	let deadline = tokio::time::Instant::now() + PAYMENT_WAIT;
+	loop {
+		let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+		let event = match tokio::time::timeout(remaining, node.next_event_async()).await {
+			Ok(event) => event,
+			Err(_) => return false,
+		};
+		node.event_handled().unwrap();
+		match event {
+			Event::PaymentSuccessful { payment_id: id, .. } if id == payment_id => return true,
+			Event::PaymentFailed { payment_id: id, .. } if id == payment_id => return false,
+			_ => {},
+		}
+	}
+}
+
+/// Calls `Node::stop` on another thread, as `stop` blocks on the runtime.
+fn stop_in_thread(node: &Arc<TestNode>) -> std::thread::JoinHandle<()> {
+	let node = Arc::clone(node);
+	std::thread::spawn(move || node.stop().unwrap())
+}
+
+/// Holds A's next `held` write while a payment over an open channel needs it, stops A, lets the
+/// write finish, restarts the same `Node` and checks that the payment and a new one settle.
+async fn do_payment_across_restart(held: HeldWrite) {
+	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();
+	let chain_source = TestChainSource::Esplora(&electrsd);
+
+	let logs = Arc::new(CollectingLogWriter::new());
+	let mut config_a = random_config();
+	config_a.log_writer = TestLogWriter::Custom(logs.clone());
+	let store = HoldingStore::new(&config_a.node_config.storage_dir_path);
+	let node_a = Arc::new(setup_node_with_store(&chain_source, config_a, store.clone()));
+	let node_b = setup_node(&chain_source, random_config());
+
+	let address_a = node_a.onchain_payment().new_address().unwrap();
+	premine_and_distribute_funds(
+		&bitcoind.client,
+		&electrsd.client,
+		vec![address_a],
+		Amount::from_sat(5_000_000),
+	)
+	.await;
+	node_a.sync_wallets().unwrap();
+	node_b.sync_wallets().unwrap();
+	// A receives when its ChannelManager write is held: then A's peer handler, not its parked
+	// background processor, starts the monitor update, which is only persisted once the
+	// background processor flushes it while A is stopping.
+	let a_receives = held == HeldWrite::ChannelManagerCommitted;
+	if a_receives {
+		open_channel_push_amt(&node_a, &node_b, 4_000_000, Some(1_500_000_000), false, &electrsd)
+			.await;
+	} else {
+		open_channel(&node_a, &node_b, 4_000_000, false, &electrsd).await;
+	}
+	generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await;
+	node_a.sync_wallets().unwrap();
+	node_b.sync_wallets().unwrap();
+	expect_channel_ready_event!(node_a, node_b.node_id());
+	expect_channel_ready_event!(node_b, node_a.node_id());
+	wait_for_channel_ready_to_send(&node_a, &node_b, 2_000_000).await;
+
+	store.hold(held);
+	let (payer, payee): (&TestNode, &TestNode) =
+		if a_receives { (&node_b, &node_a) } else { (&node_a, &node_b) };
+	let payment_id = send_payment(payer, payee, 1_000_000);
+	store.wait_until_held().await;
+
+	let stop = stop_in_thread(&node_a);
+	if a_receives {
+		// The background processor is parked on the held write, so `stop` waits for it. Only
+		// release it once `stop` has cancelled the background tasks.
+		assert!(logs.wait_for("Disconnected all network peers.").await);
+		store.release();
+		tokio::task::spawn_blocking(move || stop.join().unwrap()).await.unwrap();
+	} else {
+		// Only let the held write finish once `stop` has returned.
+		tokio::task::spawn_blocking(move || stop.join().unwrap()).await.unwrap();
+		store.release();
+	}
+
+	node_a.start().unwrap();
+	let addr_b = node_b.listening_addresses().unwrap().first().unwrap().clone();
+	let _ = node_a.connect(node_b.node_id(), addr_b, true);
+	wait_for_channel_ready_to_send(&node_a, &node_b, 600_000).await;
+	assert!(payment_succeeds(payer, payment_id).await, "payment started before stop");
+
+	let payment_id = send_payment(&node_a, &node_b, 500_000);
+	assert!(payment_succeeds(&node_a, payment_id).await, "payment after restart");
+
+	node_a.stop().unwrap();
+	node_b.stop().unwrap();
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn stop_with_monitor_update_write_in_flight() {
+	do_payment_across_restart(HeldWrite::MonitorUpdateCommitted).await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn stop_with_monitor_update_write_not_yet_issued() {
+	do_payment_across_restart(HeldWrite::MonitorUpdateUnissued).await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn stop_with_monitor_update_flushed_during_shutdown() {
+	do_payment_across_restart(HeldWrite::ChannelManagerCommitted).await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn stop_with_initial_monitor_write_in_flight() {
+	let (bitcoind, electrsd) = setup_bitcoind_and_electrsd();
+	let chain_source = TestChainSource::Esplora(&electrsd);
+
+	// B opens a channel to A, whose initial `ChannelMonitor` write is held.
+	let config_a = random_config();
+	let store = HoldingStore::new(&config_a.node_config.storage_dir_path);
+	let node_a = Arc::new(setup_node_with_store(&chain_source, config_a, store.clone()));
+	let node_b = setup_node(&chain_source, random_config());
+
+	let address_b = node_b.onchain_payment().new_address().unwrap();
+	premine_and_distribute_funds(
+		&bitcoind.client,
+		&electrsd.client,
+		vec![address_b],
+		Amount::from_sat(5_000_000),
+	)
+	.await;
+	node_a.sync_wallets().unwrap();
+	node_b.sync_wallets().unwrap();
+
+	store.hold(HeldWrite::MonitorCommitted);
+	let addr_a = node_a.listening_addresses().unwrap().first().unwrap().clone();
+	node_b.open_channel(node_a.node_id(), addr_a, 4_000_000, None, None, None).unwrap();
+	store.wait_until_held().await;
+
+	// Only let the held write finish once `stop` has returned.
+	let stop = stop_in_thread(&node_a);
+	tokio::task::spawn_blocking(move || stop.join().unwrap()).await.unwrap();
+	store.release();
+
+	node_a.start().unwrap();
+	let addr_b = node_b.listening_addresses().unwrap().first().unwrap().clone();
+	let _ = node_a.connect(node_b.node_id(), addr_b, true);
+	generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await;
+	node_a.sync_wallets().unwrap();
+	node_b.sync_wallets().unwrap();
+	wait_for_channel_ready_to_send(&node_b, &node_a, 1_000_000).await;
+
+	let payment_id = send_payment(&node_b, &node_a, 1_000_000);
+	assert!(payment_succeeds(&node_b, payment_id).await, "payment over the new channel");
+
+	node_a.stop().unwrap();
+	node_b.stop().unwrap();
+}