Uh oh!
There was an error while loading. Please reload this page.
Add support for native async KVStore persist to ChainMonitor#4063
Uh oh!
There was an error while loading. Please reload this page.
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -46,12 +46,14 @@ use crate::ln::our_peer_storage::{DecryptedOurPeerStorage, PeerStorageMonitorHol | ||
| use crate::ln::types::ChannelId; | ||
| use crate::prelude::*; | ||
| use crate::sign::ecdsa::EcdsaChannelSigner; | ||
| use crate::sign::{EntropySource, PeerStorageKey, SignerProvider}; | ||
| use crate::sync::{Mutex, MutexGuard, RwLock, RwLockReadGuard}; | ||
| use crate::types::features::{InitFeatures, NodeFeatures}; | ||
| use crate::util::async_poll::{MaybeSend, MaybeSync}; | ||
| use crate::util::errors::APIError; | ||
| use crate::util::logger::{Logger, WithContext}; | ||
| use crate::util::native_async::FutureSpawner; | ||
| use crate::util::persist::{KVStore, MonitorName, MonitorUpdatingPersisterAsync}; | ||
| #[cfg(peer_storage)] | ||
| use crate::util::ser::{VecWriter, Writeable}; | ||
| use crate::util::wakers::{Future, Notifier}; | ||
| @@ -192,6 +194,17 @@ pub trait Persist<ChannelSigner: EcdsaChannelSigner> { | ||
| /// restart, this method must in that case be idempotent, ensuring it can handle scenarios where | ||
| /// the monitor already exists in the archive. | ||
| fn archive_persisted_channel(&self, monitor_name: MonitorName); | ||
| /// Fetches the set of [`ChannelMonitorUpdate`]s, previously persisted with | ||
| /// [`Self::update_persisted_channel`], which have completed. | ||
| /// | ||
| /// Returning an update here is equivalent to calling | ||
| /// [`ChainMonitor::channel_monitor_updated`]. Because of this, this method is defaulted and | ||
| /// hidden in the docs. | ||
| #[doc(hidden)] | ||
| fn get_and_clear_completed_updates(&self) -> Vec<(ChannelId, u64)> { | ||
| Vec::new() | ||
| } | ||
| } | ||
| struct MonitorHolder<ChannelSigner: EcdsaChannelSigner> { | ||
| @@ -235,6 +248,93 @@ impl<ChannelSigner: EcdsaChannelSigner> Deref for LockedChannelMonitor<'_, Chann | ||
| } | ||
| } | ||
| /// An unconstructable [`Persist`]er which is used under the hood when you call | ||
| /// [`ChainMonitor::new_async_beta`]. | ||
| pub struct AsyncPersister< | ||
| K: Deref + MaybeSend + MaybeSync + 'static, | ||
| S: FutureSpawner, | ||
| L: Deref + MaybeSend + MaybeSync + 'static, | ||
| ES: Deref + MaybeSend + MaybeSync + 'static, | ||
| SP: Deref + MaybeSend + MaybeSync + 'static, | ||
| BI: Deref + MaybeSend + MaybeSync + 'static, | ||
| FE: Deref + MaybeSend + MaybeSync + 'static, | ||
| > where | ||
| K::Target: KVStore + MaybeSync, | ||
| L::Target: Logger, | ||
| ES::Target: EntropySource + Sized, | ||
| SP::Target: SignerProvider + Sized, | ||
| BI::Target: BroadcasterInterface, | ||
| FE::Target: FeeEstimator, | ||
| { | ||
| persister: MonitorUpdatingPersisterAsync<K, S, L, ES, SP, BI, FE>, | ||
| } | ||
| impl< | ||
| K: Deref + MaybeSend + MaybeSync + 'static, | ||
| S: FutureSpawner, | ||
| L: Deref + MaybeSend + MaybeSync + 'static, | ||
| ES: Deref + MaybeSend + MaybeSync + 'static, | ||
| SP: Deref + MaybeSend + MaybeSync + 'static, | ||
| BI: Deref + MaybeSend + MaybeSync + 'static, | ||
| FE: Deref + MaybeSend + MaybeSync + 'static, | ||
| > Deref for AsyncPersister<K, S, L, ES, SP, BI, FE> | ||
| where | ||
| K::Target: KVStore + MaybeSync, | ||
| L::Target: Logger, | ||
| ES::Target: EntropySource + Sized, | ||
| SP::Target: SignerProvider + Sized, | ||
| BI::Target: BroadcasterInterface, | ||
| FE::Target: FeeEstimator, | ||
| { | ||
| type Target = Self; | ||
| fn deref(&self) -> &Self { | ||
| self | ||
| } | ||
| } | ||
| impl< | ||
| K: Deref + MaybeSend + MaybeSync + 'static, | ||
| S: FutureSpawner, | ||
| L: Deref + MaybeSend + MaybeSync + 'static, | ||
| ES: Deref + MaybeSend + MaybeSync + 'static, | ||
| SP: Deref + MaybeSend + MaybeSync + 'static, | ||
| BI: Deref + MaybeSend + MaybeSync + 'static, | ||
| FE: Deref + MaybeSend + MaybeSync + 'static, | ||
| > Persist<<SP::Target as SignerProvider>::EcdsaSigner> for AsyncPersister<K, S, L, ES, SP, BI, FE> | ||
| where | ||
| K::Target: KVStore + MaybeSync, | ||
| L::Target: Logger, | ||
| ES::Target: EntropySource + Sized, | ||
| SP::Target: SignerProvider + Sized, | ||
| BI::Target: BroadcasterInterface, | ||
| FE::Target: FeeEstimator, | ||
| <SP::Target as SignerProvider>::EcdsaSigner: MaybeSend + 'static, | ||
| { | ||
| fn persist_new_channel( | ||
| &self, monitor_name: MonitorName, | ||
| monitor: &ChannelMonitor<<SP::Target as SignerProvider>::EcdsaSigner>, | ||
| ) -> ChannelMonitorUpdateStatus { | ||
| self.persister.spawn_async_persist_new_channel(monitor_name, monitor); | ||
| ChannelMonitorUpdateStatus::InProgress | ||
| } | ||
| fn update_persisted_channel( | ||
| &self, monitor_name: MonitorName, monitor_update: Option<&ChannelMonitorUpdate>, | ||
| monitor: &ChannelMonitor<<SP::Target as SignerProvider>::EcdsaSigner>, | ||
| ) -> ChannelMonitorUpdateStatus { | ||
| self.persister.spawn_async_update_persisted_channel(monitor_name, monitor_update, monitor); | ||
| ChannelMonitorUpdateStatus::InProgress | ||
| } | ||
| fn archive_persisted_channel(&self, monitor_name: MonitorName) { | ||
| self.persister.spawn_async_archive_persisted_channel(monitor_name); | ||
| } | ||
| fn get_and_clear_completed_updates(&self) -> Vec<(ChannelId, u64)> { | ||
| self.persister.get_and_clear_completed_updates() | ||
| } | ||
| } | ||
| /// An implementation of [`chain::Watch`] for monitoring channels. | ||
| /// | ||
| /// Connected and disconnected blocks must be provided to `ChainMonitor` as documented by | ||
| @@ -291,6 +391,63 @@ pub struct ChainMonitor< | ||
| our_peerstorage_encryption_key: PeerStorageKey, | ||
| } | ||
| impl< | ||
| K: Deref + MaybeSend + MaybeSync + 'static, | ||
| S: FutureSpawner, | ||
| SP: Deref + MaybeSend + MaybeSync + 'static, | ||
| C: Deref, | ||
| T: Deref + MaybeSend + MaybeSync + 'static, | ||
| F: Deref + MaybeSend + MaybeSync + 'static, | ||
| L: Deref + MaybeSend + MaybeSync + 'static, | ||
| ES: Deref + MaybeSend + MaybeSync + 'static, | ||
| > | ||
| ChainMonitor< | ||
| <SP::Target as SignerProvider>::EcdsaSigner, | ||
| C, | ||
| T, | ||
| F, | ||
| L, | ||
| AsyncPersister<K, S, L, ES, SP, T, F>, | ||
| ES, | ||
| > where | ||
| K::Target: KVStore + MaybeSync, | ||
| SP::Target: SignerProvider + Sized, | ||
| C::Target: chain::Filter, | ||
| T::Target: BroadcasterInterface, | ||
| F::Target: FeeEstimator, | ||
| L::Target: Logger, | ||
| ES::Target: EntropySource + Sized, | ||
| <SP::Target as SignerProvider>::EcdsaSigner: MaybeSend + 'static, | ||
| { | ||
| /// Creates a new `ChainMonitor` used to watch on-chain activity pertaining to channels. | ||
| /// | ||
| /// This behaves the same as [`ChainMonitor::new`] except that it relies on | ||
| /// [`MonitorUpdatingPersisterAsync`] and thus allows persistence to be completed async. | ||
| /// | ||
| /// Note that async monitor updating is considered beta, and bugs may be triggered by its use. | ||
| pub fn new_async_beta( | ||
| chain_source: Option<C>, broadcaster: T, logger: L, feeest: F, | ||
Contributor There was a problem hiding this comment. nit: Why not simply CollaboratorAuthor There was a problem hiding this comment. Cause I copied the one from | ||
| persister: MonitorUpdatingPersisterAsync<K, S, L, ES, SP, T, F>, _entropy_source: ES, | ||
| _our_peerstorage_encryption_key: PeerStorageKey, | ||
| ) -> Self { | ||
| Self { | ||
| monitors: RwLock::new(new_hash_map()), | ||
| chain_source, | ||
| broadcaster, | ||
| logger, | ||
| fee_estimator: feeest, | ||
| persister: AsyncPersister { persister }, | ||
| _entropy_source, | ||
| pending_monitor_events: Mutex::new(Vec::new()), | ||
| highest_chain_height: AtomicUsize::new(0), | ||
| event_notifier: Notifier::new(), | ||
| pending_send_only_events: Mutex::new(Vec::new()), | ||
| #[cfg(peer_storage)] | ||
| our_peerstorage_encryption_key: _our_peerstorage_encryption_key, | ||
| } | ||
| } | ||
| } | ||
| impl< | ||
| ChannelSigner: EcdsaChannelSigner, | ||
| C: Deref, | ||
| @@ -1357,6 +1514,9 @@ where | ||
| fn release_pending_monitor_events( | ||
| &self, | ||
| ) -> Vec<(OutPoint, ChannelId, Vec<MonitorEvent>, PublicKey)> { | ||
| for (channel_id, update_id) in self.persister.get_and_clear_completed_updates() { | ||
joostjager marked this conversation as resolved.
Uh oh!There was an error while loading. Please reload this page. | ||
| let _ = self.channel_monitor_updated(channel_id, update_id); | ||
| } | ||
| let mut pending_monitor_events = self.pending_monitor_events.lock().unwrap().split_off(0); | ||
| for monitor_state in self.monitors.read().unwrap().values() { | ||
| let monitor_events = monitor_state.monitor.get_and_clear_pending_monitor_events(); | ||
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think there are now three ways to do persistence: sync, the previous async way via implementing a different
Persistand thisnew_async_beta?Is there any form of consolidation possible between the two async setups?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yea, I mention it in the last commit, but I think the eventual consolidation should be that we merge
MonitorUpdatingPersisterintoChainMonitorand then thePersistinterface is just the interface betweenChannelManagerandChainMonitor, a user will always just instantiate aChainMonitorwith either aKVStoreor aKVStoreSyncand we'll deal with the rest.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
That makes sense to me. I just wondered if we should already now steer towards
MonitorUpdatingPersisterwith an async kv store as the only way to do async. I don't think it is more "beta" than the current callback-based async?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thoughts on this?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ah, I'd missed this. I don't really see a strong reason to change the current API and remove the manual-async approach immediately. Its not additional code to maintain (given the new logic uses it under the hood anyway) and we do have folks using it. That said, it does probably make sense to deprecate it, which I'll go ahead and do here.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Oh actually nevermind, we should have a discussion about if we want to support async outside of rust, which would need the old API (or a way to make async KVStore work outside of rust, which I think we can do eventually as well).
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I didn't consider the bindings. Not great to remain stuck with multiple ways to do it, but not sure what we can do either.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think we can map the async stuff to bindings eventually, so its not like we're stuck, just a question of priorities.