Drop required `Box`ing of `KVStore` `Future`s

diff --git a/ci/check-lint.sh b/ci/check-lint.sh
index 39c1069..c1f1b08 100755
--- a/ci/check-lint.sh
+++ b/ci/check-lint.sh
@@ -107,7 +107,8 @@ CLIPPY() {
 		-A clippy::useless_conversion \
 		-A clippy::manual_repeat_n `# to be removed once we hit MSRV 1.86` \
 		-A clippy::manual_is_multiple_of `# to be removed once we hit MSRV 1.87` \
-		-A clippy::uninlined-format-args
+		-A clippy::uninlined-format-args \
+		-A clippy::manual-async-fn # Not really sure why this is even a warning when there's a Send bound
 }
 
 CLIPPY
diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs
index 19333c5..bc0d42a 100644
--- a/lightning-background-processor/src/lib.rs
+++ b/lightning-background-processor/src/lib.rs
@@ -41,6 +41,8 @@ use lightning::events::ReplayEvent;
 use lightning::events::{Event, PathFailure};
 use lightning::util::ser::Writeable;
 
+#[cfg(not(c_bindings))]
+use lightning::io::Error;
 use lightning::ln::channelmanager::AChannelManager;
 use lightning::ln::msgs::OnionMessageHandler;
 use lightning::ln::peer_handler::APeerManager;
@@ -51,6 +53,8 @@ use lightning::routing::utxo::UtxoLookup;
 use lightning::sign::{
 	ChangeDestinationSource, ChangeDestinationSourceSync, EntropySource, OutputSpender,
 };
+#[cfg(not(c_bindings))]
+use lightning::util::async_poll::MaybeSend;
 use lightning::util::logger::Logger;
 use lightning::util::persist::{
 	KVStore, KVStoreSync, KVStoreSyncWrapper, CHANNEL_MANAGER_PERSISTENCE_KEY,
@@ -83,7 +87,11 @@ use std::time::Instant;
 #[cfg(not(feature = "std"))]
 use alloc::boxed::Box;
 #[cfg(all(not(c_bindings), not(feature = "std")))]
+use alloc::string::String;
+#[cfg(all(not(c_bindings), not(feature = "std")))]
 use alloc::sync::Arc;
+#[cfg(all(not(c_bindings), not(feature = "std")))]
+use alloc::vec::Vec;
 
 /// `BackgroundProcessor` takes care of tasks that (1) need to happen periodically to keep
 /// Rust-Lightning running properly, and (2) either can or should be run in the background. Its
@@ -416,6 +424,37 @@ pub const NO_ONION_MESSENGER: Option<
 	>,
 > = None;
 
+#[cfg(not(c_bindings))]
+/// A panicking implementation of [`KVStore`] that is used in [`NO_LIQUIDITY_MANAGER`].
+pub struct DummyKVStore;
+
+#[cfg(not(c_bindings))]
+impl KVStore for DummyKVStore {
+	fn read(
+		&self, _: &str, _: &str, _: &str,
+	) -> impl core::future::Future<Output = Result<Vec<u8>, Error>> + MaybeSend + 'static {
+		async { unimplemented!() }
+	}
+
+	fn write(
+		&self, _: &str, _: &str, _: &str, _: Vec<u8>,
+	) -> impl core::future::Future<Output = Result<(), Error>> + MaybeSend + 'static {
+		async { unimplemented!() }
+	}
+
+	fn remove(
+		&self, _: &str, _: &str, _: &str, _: bool,
+	) -> impl core::future::Future<Output = Result<(), Error>> + MaybeSend + 'static {
+		async { unimplemented!() }
+	}
+
+	fn list(
+		&self, _: &str, _: &str,
+	) -> impl core::future::Future<Output = Result<Vec<String>, Error>> + MaybeSend + 'static {
+		async { unimplemented!() }
+	}
+}
+
 /// When initializing a background processor without a liquidity manager, this can be used to avoid
 /// specifying a concrete `LiquidityManager` type.
 #[cfg(not(c_bindings))]
@@ -430,8 +469,8 @@ pub const NO_LIQUIDITY_MANAGER: Option<
 				CM = &DynChannelManager,
 				Filter = dyn chain::Filter + Send + Sync,
 				C = &(dyn chain::Filter + Send + Sync),
-				KVStore = dyn lightning::util::persist::KVStore + Send + Sync,
-				K = &(dyn lightning::util::persist::KVStore + Send + Sync),
+				KVStore = DummyKVStore,
+				K = &DummyKVStore,
 				TimeProvider = dyn lightning_liquidity::utils::time::TimeProvider + Send + Sync,
 				TP = &(dyn lightning_liquidity::utils::time::TimeProvider + Send + Sync),
 				BroadcasterInterface = dyn lightning::chain::chaininterface::BroadcasterInterface
diff --git a/lightning-persister/src/fs_store.rs b/lightning-persister/src/fs_store.rs
index 9b15398..b2d327f 100644
--- a/lightning-persister/src/fs_store.rs
+++ b/lightning-persister/src/fs_store.rs
@@ -14,8 +14,6 @@ use std::sync::{Arc, Mutex, RwLock};
 #[cfg(feature = "tokio")]
 use core::future::Future;
 #[cfg(feature = "tokio")]
-use core::pin::Pin;
-#[cfg(feature = "tokio")]
 use lightning::util::persist::KVStore;
 
 #[cfg(target_os = "windows")]
@@ -464,93 +462,85 @@ impl FilesystemStoreInner {
 impl KVStore for FilesystemStore {
 	fn read(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
-	) -> Pin<Box<dyn Future<Output = Result<Vec<u8>, lightning::io::Error>> + 'static + Send>> {
+	) -> impl Future<Output = Result<Vec<u8>, lightning::io::Error>> + 'static + Send {
 		let this = Arc::clone(&self.inner);
-		let path = match this.get_checked_dest_file_path(
+		let path = this.get_checked_dest_file_path(
 			primary_namespace,
 			secondary_namespace,
 			Some(key),
 			"read",
-		) {
-			Ok(path) => path,
-			Err(e) => return Box::pin(async move { Err(e) }),
-		};
+		);
 
-		Box::pin(async move {
+		async move {
+			let path = match path {
+				Ok(path) => path,
+				Err(e) => return Err(e),
+			};
 			tokio::task::spawn_blocking(move || this.read(path)).await.unwrap_or_else(|e| {
 				Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, e))
 			})
-		})
+		}
 	}
 
 	fn write(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
-	) -> Pin<Box<dyn Future<Output = Result<(), lightning::io::Error>> + 'static + Send>> {
+	) -> impl Future<Output = Result<(), lightning::io::Error>> + 'static + Send {
 		let this = Arc::clone(&self.inner);
-		let path = match this.get_checked_dest_file_path(
-			primary_namespace,
-			secondary_namespace,
-			Some(key),
-			"write",
-		) {
-			Ok(path) => path,
-			Err(e) => return Box::pin(async move { Err(e) }),
-		};
-
-		let (inner_lock_ref, version) = self.get_new_version_and_lock_ref(path.clone());
-		Box::pin(async move {
+		let path = this
+			.get_checked_dest_file_path(primary_namespace, secondary_namespace, Some(key), "write")
+			.map(|path| (self.get_new_version_and_lock_ref(path.clone()), path));
+
+		async move {
+			let ((inner_lock_ref, version), path) = match path {
+				Ok(res) => res,
+				Err(e) => return Err(e),
+			};
 			tokio::task::spawn_blocking(move || {
 				this.write_version(inner_lock_ref, path, buf, version)
 			})
 			.await
 			.unwrap_or_else(|e| Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, e)))
-		})
+		}
 	}
 
 	fn remove(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
-	) -> Pin<Box<dyn Future<Output = Result<(), lightning::io::Error>> + 'static + Send>> {
+	) -> impl Future<Output = Result<(), lightning::io::Error>> + 'static + Send {
 		let this = Arc::clone(&self.inner);
-		let path = match this.get_checked_dest_file_path(
-			primary_namespace,
-			secondary_namespace,
-			Some(key),
-			"remove",
-		) {
-			Ok(path) => path,
-			Err(e) => return Box::pin(async move { Err(e) }),
-		};
-
-		let (inner_lock_ref, version) = self.get_new_version_and_lock_ref(path.clone());
-		Box::pin(async move {
+		let path = this
+			.get_checked_dest_file_path(primary_namespace, secondary_namespace, Some(key), "remove")
+			.map(|path| (self.get_new_version_and_lock_ref(path.clone()), path));
+
+		async move {
+			let ((inner_lock_ref, version), path) = match path {
+				Ok(res) => res,
+				Err(e) => return Err(e),
+			};
 			tokio::task::spawn_blocking(move || {
 				this.remove_version(inner_lock_ref, path, lazy, version)
 			})
 			.await
 			.unwrap_or_else(|e| Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, e)))
-		})
+		}
 	}
 
 	fn list(
 		&self, primary_namespace: &str, secondary_namespace: &str,
-	) -> Pin<Box<dyn Future<Output = Result<Vec<String>, lightning::io::Error>> + 'static + Send>> {
+	) -> impl Future<Output = Result<Vec<String>, lightning::io::Error>> + 'static + Send {
 		let this = Arc::clone(&self.inner);
 
-		let path = match this.get_checked_dest_file_path(
-			primary_namespace,
-			secondary_namespace,
-			None,
-			"list",
-		) {
-			Ok(path) => path,
-			Err(e) => return Box::pin(async move { Err(e) }),
-		};
+		let path =
+			this.get_checked_dest_file_path(primary_namespace, secondary_namespace, None, "list");
 
-		Box::pin(async move {
+		async move {
+			let path = match path {
+				Ok(path) => path,
+				Err(e) => return Err(e),
+			};
 			tokio::task::spawn_blocking(move || this.list(path)).await.unwrap_or_else(|e| {
 				Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, e))
 			})
-		})
+		}
 	}
 }
 
@@ -758,24 +748,24 @@ mod tests {
 		let fs_store = Arc::new(FilesystemStore::new(temp_path));
 		assert_eq!(fs_store.state_size(), 0);
 
-		let async_fs_store: Arc<dyn KVStore> = fs_store.clone();
+		let async_fs_store = Arc::clone(&fs_store);
 
 		let data1 = vec![42u8; 32];
 		let data2 = vec![43u8; 32];
 
-		let primary_namespace = "testspace";
-		let secondary_namespace = "testsubspace";
+		let primary = "testspace";
+		let secondary = "testsubspace";
 		let key = "testkey";
 
 		// Test writing the same key twice with different data. Execute the asynchronous part out of order to ensure
 		// that eventual consistency works.
-		let fut1 = async_fs_store.write(primary_namespace, secondary_namespace, key, data1);
+		let fut1 = KVStore::write(&*async_fs_store, primary, secondary, key, data1);
 		assert_eq!(fs_store.state_size(), 1);
 
-		let fut2 = async_fs_store.remove(primary_namespace, secondary_namespace, key, false);
+		let fut2 = KVStore::remove(&*async_fs_store, primary, secondary, key, false);
 		assert_eq!(fs_store.state_size(), 1);
 
-		let fut3 = async_fs_store.write(primary_namespace, secondary_namespace, key, data2.clone());
+		let fut3 = KVStore::write(&*async_fs_store, primary, secondary, key, data2.clone());
 		assert_eq!(fs_store.state_size(), 1);
 
 		fut3.await.unwrap();
@@ -788,21 +778,18 @@ mod tests {
 		assert_eq!(fs_store.state_size(), 0);
 
 		// Test list.
-		let listed_keys =
-			async_fs_store.list(primary_namespace, secondary_namespace).await.unwrap();
+		let listed_keys = KVStore::list(&*async_fs_store, primary, secondary).await.unwrap();
 		assert_eq!(listed_keys.len(), 1);
 		assert_eq!(listed_keys[0], key);
 
 		// Test read. We expect to read data2, as the write call was initiated later.
-		let read_data =
-			async_fs_store.read(primary_namespace, secondary_namespace, key).await.unwrap();
+		let read_data = KVStore::read(&*async_fs_store, primary, secondary, key).await.unwrap();
 		assert_eq!(data2, &*read_data);
 
 		// Test remove.
-		async_fs_store.remove(primary_namespace, secondary_namespace, key, false).await.unwrap();
+		KVStore::remove(&*async_fs_store, primary, secondary, key, false).await.unwrap();
 
-		let listed_keys =
-			async_fs_store.list(primary_namespace, secondary_namespace).await.unwrap();
+		let listed_keys = KVStore::list(&*async_fs_store, primary, secondary).await.unwrap();
 		assert_eq!(listed_keys.len(), 0);
 	}
 
diff --git a/lightning/src/util/persist.rs b/lightning/src/util/persist.rs
index 3ad9b42..7feb781 100644
--- a/lightning/src/util/persist.rs
+++ b/lightning/src/util/persist.rs
@@ -34,7 +34,7 @@ use crate::chain::transaction::OutPoint;
 use crate::ln::types::ChannelId;
 use crate::sign::{ecdsa::EcdsaChannelSigner, EntropySource, SignerProvider};
 use crate::sync::Mutex;
-use crate::util::async_poll::{dummy_waker, AsyncResult, MaybeSend, MaybeSync};
+use crate::util::async_poll::{dummy_waker, MaybeSend, MaybeSync};
 use crate::util::logger::Logger;
 use crate::util::native_async::FutureSpawner;
 use crate::util::ser::{Readable, ReadableArgs, Writeable};
@@ -216,34 +216,34 @@ where
 {
 	fn read(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
-	) -> AsyncResult<'static, Vec<u8>, io::Error> {
+	) -> impl Future<Output = Result<Vec<u8>, io::Error>> + 'static + MaybeSend {
 		let res = self.0.read(primary_namespace, secondary_namespace, key);
 
-		Box::pin(async move { res })
+		async move { res }
 	}
 
 	fn write(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
-	) -> AsyncResult<'static, (), io::Error> {
+	) -> impl Future<Output = Result<(), io::Error>> + 'static + MaybeSend {
 		let res = self.0.write(primary_namespace, secondary_namespace, key, buf);
 
-		Box::pin(async move { res })
+		async move { res }
 	}
 
 	fn remove(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
-	) -> AsyncResult<'static, (), io::Error> {
+	) -> impl Future<Output = Result<(), io::Error>> + 'static + MaybeSend {
 		let res = self.0.remove(primary_namespace, secondary_namespace, key, lazy);
 
-		Box::pin(async move { res })
+		async move { res }
 	}
 
 	fn list(
 		&self, primary_namespace: &str, secondary_namespace: &str,
-	) -> AsyncResult<'static, Vec<String>, io::Error> {
+	) -> impl Future<Output = Result<Vec<String>, io::Error>> + 'static + MaybeSend {
 		let res = self.0.list(primary_namespace, secondary_namespace);
 
-		Box::pin(async move { res })
+		async move { res }
 	}
 }
 
@@ -283,16 +283,18 @@ pub trait KVStore {
 	/// [`ErrorKind::NotFound`]: io::ErrorKind::NotFound
 	fn read(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
-	) -> AsyncResult<'static, Vec<u8>, io::Error>;
+	) -> impl Future<Output = Result<Vec<u8>, io::Error>> + 'static + MaybeSend;
 	/// Persists the given data under the given `key`.
 	///
-	/// The order of multiple writes to the same key needs to be retained while persisting
-	/// asynchronously. In other words, if two writes to the same key occur, the state (as seen by
-	/// [`Self::read`]) must either see the first write then the second, or only ever the second,
-	/// no matter when the futures complete (and must always contain the second write once the
-	/// second future completes). The state should never contain the first write after the second
-	/// write's future completes, nor should it contain the second write, then contain the first
-	/// write at any point thereafter (even if the second write's future hasn't yet completed).
+	/// Note that this is *not* an `async fn`. Rather, the order of multiple writes to the same key
+	/// (as defined by the order of the synchronous function calls) needs to be retained while
+	/// persisting asynchronously. In other words, if two writes to the same key occur, the state
+	/// (as seen by [`Self::read`]) must either see the first write then the second, or only ever
+	/// the second, no matter when the futures complete (and must always contain the second write
+	/// once the second future completes). The state should never contain the first write after the
+	/// second write's future completes, nor should it contain the second write, then contain the
+	/// first write at any point thereafter (even if the second write's future hasn't yet
+	/// completed).
 	///
 	/// One way to ensure this requirement is met is by assigning a version number to each write
 	/// before returning the future, and then during asynchronous execution, ensuring that the
@@ -303,7 +305,7 @@ pub trait KVStore {
 	/// Will create the given `primary_namespace` and `secondary_namespace` if not already present in the store.
 	fn write(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
-	) -> AsyncResult<'static, (), io::Error>;
+	) -> impl Future<Output = Result<(), io::Error>> + 'static + MaybeSend;
 	/// Removes any data that had previously been persisted under the given `key`.
 	///
 	/// If the `lazy` flag is set to `true`, the backend implementation might choose to lazily
@@ -311,6 +313,10 @@ pub trait KVStore {
 	/// eventual batch deletion of multiple keys. As a consequence, subsequent calls to
 	/// [`KVStoreSync::list`] might include the removed key until the changes are actually persisted.
 	///
+	/// Note that similar to [`Self::write`] this is *not* an `async fn`, but rather a sync fn
+	/// which defines the order of writes to a given key, but which may complete its operation
+	/// asynchronously.
+	///
 	/// Note that while setting the `lazy` flag reduces the I/O burden of multiple subsequent
 	/// `remove` calls, it also influences the atomicity guarantees as lazy `remove`s could
 	/// potentially get lost on crash after the method returns. Therefore, this flag should only be
@@ -321,12 +327,13 @@ pub trait KVStore {
 	/// to the same key which occur before a removal completes must cancel/overwrite the pending
 	/// removal.
 	///
+	///
 	/// Returns successfully if no data will be stored for the given `primary_namespace`,
 	/// `secondary_namespace`, and `key`, independently of whether it was present before its
 	/// invokation or not.
 	fn remove(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
-	) -> AsyncResult<'static, (), io::Error>;
+	) -> impl Future<Output = Result<(), io::Error>> + 'static + MaybeSend;
 	/// Returns a list of keys that are stored under the given `secondary_namespace` in
 	/// `primary_namespace`.
 	///
@@ -334,7 +341,7 @@ pub trait KVStore {
 	/// returned keys. Returns an empty list if `primary_namespace` or `secondary_namespace` is unknown.
 	fn list(
 		&self, primary_namespace: &str, secondary_namespace: &str,
-	) -> AsyncResult<'static, Vec<String>, io::Error>;
+	) -> impl Future<Output = Result<Vec<String>, io::Error>> + 'static + MaybeSend;
 }
 
 /// Provides additional interface methods that are required for [`KVStore`]-to-[`KVStore`]
@@ -1005,6 +1012,9 @@ where
 	}
 }
 
+trait MaybeSendableFuture: Future<Output = Result<(), io::Error>> + MaybeSend {}
+impl<F: Future<Output = Result<(), io::Error>> + MaybeSend> MaybeSendableFuture for F {}
+
 impl<K: Deref, S: FutureSpawner, L: Deref, ES: Deref, SP: Deref, BI: Deref, FE: Deref>
 	MonitorUpdatingPersisterAsyncInner<K, S, L, ES, SP, BI, FE>
 where
@@ -1178,9 +1188,9 @@ where
 		Ok(())
 	}
 
-	fn persist_new_channel<ChannelSigner: EcdsaChannelSigner>(
-		&self, monitor_name: MonitorName, monitor: &ChannelMonitor<ChannelSigner>,
-	) -> impl Future<Output = Result<(), io::Error>> {
+	fn persist_new_channel<'a, ChannelSigner: EcdsaChannelSigner>(
+		&'a self, monitor_name: MonitorName, monitor: &'a ChannelMonitor<ChannelSigner>,
+	) -> Pin<Box<dyn MaybeSendableFuture<Output = Result<(), io::Error>> + 'static>> {
 		// Determine the proper key for this monitor
 		let monitor_key = monitor_name.to_string();
 		// Serialize and write the new monitor
@@ -1199,7 +1209,10 @@ where
 		// completion of the write. This ensures monitor persistence ordering is preserved.
 		let primary = CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE;
 		let secondary = CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE;
-		self.kv_store.write(primary, secondary, monitor_key.as_str(), monitor_bytes)
+		// There's no real reason why this needs to be boxed, but dropping it rams into the "hidden
+		// type for impl... captures lifetime that does not appear in bounds" issue. This can
+		// trivially be dropped once we upgrade to edition 2024/MSRV 1.85.
+		Box::pin(self.kv_store.write(primary, secondary, monitor_key.as_str(), monitor_bytes))
 	}
 
 	fn update_persisted_channel<'a, ChannelSigner: EcdsaChannelSigner + 'a>(
@@ -1225,12 +1238,10 @@ where
 				// write method, allowing it to do its queueing immediately, and then return a
 				// future for the completion of the write. This ensures monitor persistence
 				// ordering is preserved.
-				res_a = Some(self.kv_store.write(
-					primary,
-					&monitor_key,
-					update_name.as_str(),
-					update.encode(),
-				));
+				let encoded = update.encode();
+				res_a = Some(async move {
+					self.kv_store.write(primary, &monitor_key, update_name.as_str(), encoded).await
+				});
 			} else {
 				// We could write this update, but it meets criteria of our design that calls for a full monitor write.
 				// Note that this is NOT an async function, but rather calls the *sync* KVStore
diff --git a/lightning/src/util/sweep.rs b/lightning/src/util/sweep.rs
index a3ded6f..bf048ef 100644
--- a/lightning/src/util/sweep.rs
+++ b/lightning/src/util/sweep.rs
@@ -35,11 +35,11 @@ use bitcoin::{BlockHash, ScriptBuf, Transaction, Txid};
 
 use core::future::Future;
 use core::ops::Deref;
-use core::pin::pin;
+use core::pin::{pin, Pin};
 use core::sync::atomic::{AtomicBool, Ordering};
 use core::task;
 
-use super::async_poll::{dummy_waker, AsyncResult};
+use super::async_poll::dummy_waker;
 
 /// The number of blocks we wait before we prune the tracked spendable outputs.
 pub const PRUNE_DELAY_BLOCKS: u32 = ARCHIVAL_DELAY_BLOCKS + ANTI_REORG_DELAY;
@@ -610,15 +610,32 @@ where
 		sweeper_state.dirty = true;
 	}
 
-	fn persist_state<'a>(&self, sweeper_state: &SweeperState) -> AsyncResult<'a, (), io::Error> {
+	#[cfg(feature = "std")]
+	fn persist_state<'a>(
+		&'a self, sweeper_state: &SweeperState,
+	) -> Pin<Box<dyn Future<Output = Result<(), io::Error>> + Send + 'static>> {
 		let encoded = sweeper_state.encode();
 
-		self.kv_store.write(
+		Box::pin(self.kv_store.write(
 			OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE,
 			OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE,
 			OUTPUT_SWEEPER_PERSISTENCE_KEY,
 			encoded,
-		)
+		))
+	}
+
+	#[cfg(not(feature = "std"))]
+	fn persist_state<'a>(
+		&'a self, sweeper_state: &SweeperState,
+	) -> Pin<Box<dyn Future<Output = Result<(), io::Error>> + 'static>> {
+		let encoded = sweeper_state.encode();
+
+		Box::pin(self.kv_store.write(
+			OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE,
+			OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE,
+			OUTPUT_SWEEPER_PERSISTENCE_KEY,
+			encoded,
+		))
 	}
 
 	/// Updates the sweeper state by executing the given callback. Persists the state afterwards if it is marked dirty,
diff --git a/lightning/src/util/test_utils.rs b/lightning/src/util/test_utils.rs
index ad8ea22..b4db17b 100644
--- a/lightning/src/util/test_utils.rs
+++ b/lightning/src/util/test_utils.rs
@@ -50,7 +50,7 @@ use crate::sign::{self, ReceiveAuthKey};
 use crate::sign::{ChannelSigner, PeerStorageKey};
 use crate::sync::RwLock;
 use crate::types::features::{ChannelFeatures, InitFeatures, NodeFeatures};
-use crate::util::async_poll::AsyncResult;
+use crate::util::async_poll::MaybeSend;
 use crate::util::config::UserConfig;
 use crate::util::dyn_signer::{
 	DynKeysInterface, DynKeysInterfaceTrait, DynPhantomKeysInterface, DynSigner,
@@ -1012,13 +1012,13 @@ impl TestStore {
 impl KVStore for TestStore {
 	fn read(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
-	) -> AsyncResult<'static, Vec<u8>, io::Error> {
+	) -> impl Future<Output = Result<Vec<u8>, io::Error>> + 'static + MaybeSend {
 		let res = self.read_internal(&primary_namespace, &secondary_namespace, &key);
-		Box::pin(async move { res })
+		async move { res }
 	}
 	fn write(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
-	) -> AsyncResult<'static, (), io::Error> {
+	) -> impl Future<Output = Result<(), io::Error>> + 'static + MaybeSend {
 		let path = format!("{primary_namespace}/{secondary_namespace}/{key}");
 		let future = Arc::new(Mutex::new((None, None)));
 
@@ -1027,19 +1027,19 @@ impl KVStore for TestStore {
 		let new_id = pending_writes.last().map(|(id, _, _)| id + 1).unwrap_or(0);
 		pending_writes.push((new_id, Arc::clone(&future), buf));
 
-		Box::pin(OneShotChannel(future))
+		OneShotChannel(future)
 	}
 	fn remove(
 		&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
-	) -> AsyncResult<'static, (), io::Error> {
+	) -> impl Future<Output = Result<(), io::Error>> + 'static + MaybeSend {
 		let res = self.remove_internal(&primary_namespace, &secondary_namespace, &key, lazy);
-		Box::pin(async move { res })
+		async move { res }
 	}
 	fn list(
 		&self, primary_namespace: &str, secondary_namespace: &str,
-	) -> AsyncResult<'static, Vec<String>, io::Error> {
+	) -> impl Future<Output = Result<Vec<String>, io::Error>> + 'static + MaybeSend {
 		let res = self.list_internal(primary_namespace, secondary_namespace);
-		Box::pin(async move { res })
+		async move { res }
 	}
 }