Merge pull request #2964 from jbesraa/prune-stale-chanmonitor
[rust-lightning] / lightning / src / util / persist.rs
index 35f5b0c79224b18fb4b1aa30a6cad2bdd585a49f..249a089cd4883170be76bf8574855379ab48c643 100644 (file)
 //! [`ChannelManager`]: crate::ln::channelmanager::ChannelManager
 
 use core::cmp;
-use core::convert::{TryFrom, TryInto};
 use core::ops::Deref;
 use core::str::FromStr;
 use bitcoin::{BlockHash, Txid};
 
 use crate::{io, log_error};
-use crate::alloc::string::ToString;
 use crate::prelude::*;
 
 use crate::chain;
@@ -58,6 +56,11 @@ pub const CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";
 /// The primary namespace under which [`ChannelMonitorUpdate`]s will be persisted.
 pub const CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE: &str = "monitor_updates";
 
+/// The primary namespace under which archived [`ChannelMonitor`]s will be persisted.
+pub const ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE: &str = "archived_monitors";
+/// The secondary namespace under which archived [`ChannelMonitor`]s will be persisted.
+pub const ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";
+
 /// The primary namespace under which the [`NetworkGraph`] will be persisted.
 pub const NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE: &str = "";
 /// The secondary namespace under which the [`NetworkGraph`] will be persisted.
@@ -72,6 +75,20 @@ pub const SCORER_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";
 /// The key under which the [`WriteableScore`] will be persisted.
 pub const SCORER_PERSISTENCE_KEY: &str = "scorer";
 
+/// The primary namespace under which [`OutputSweeper`] state will be persisted.
+///
+/// [`OutputSweeper`]: crate::util::sweep::OutputSweeper
+pub const OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE: &str = "";
+/// The secondary namespace under which [`OutputSweeper`] state will be persisted.
+///
+/// [`OutputSweeper`]: crate::util::sweep::OutputSweeper
+pub const OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";
+/// The secondary namespace under which [`OutputSweeper`] state will be persisted.
+/// The key under which [`OutputSweeper`] state will be persisted.
+///
+/// [`OutputSweeper`]: crate::util::sweep::OutputSweeper
+pub const OUTPUT_SWEEPER_PERSISTENCE_KEY: &str = "output_sweeper";
+
 /// A sentinel value to be prepended to monitors persisted by the [`MonitorUpdatingPersister`].
 ///
 /// This serves to prevent someone from accidentally loading such monitors (which may need
@@ -158,34 +175,7 @@ where
 }
 
 
-impl<'a, A: KVStore, CM: Deref, L: Deref, S: WriteableScore<'a>> Persister<'a, CM, L, S> for A
-where
-       CM::Target: 'static + AChannelManager,
-       L::Target: 'static + Logger,
-{
-       fn persist_manager(&self, channel_manager: &CM) -> Result<(), io::Error> {
-               self.write(CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
-                       CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
-                       CHANNEL_MANAGER_PERSISTENCE_KEY,
-                       &channel_manager.get_cm().encode())
-       }
-
-       fn persist_graph(&self, network_graph: &NetworkGraph<L>) -> Result<(), io::Error> {
-               self.write(NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
-                       NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE,
-                       NETWORK_GRAPH_PERSISTENCE_KEY,
-                       &network_graph.encode())
-       }
-
-       fn persist_scorer(&self, scorer: &S) -> Result<(), io::Error> {
-               self.write(SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
-                       SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
-                       SCORER_PERSISTENCE_KEY,
-                       &scorer.encode())
-       }
-}
-
-impl<'a, CM: Deref, L: Deref, S: WriteableScore<'a>> Persister<'a, CM, L, S> for dyn KVStore + Send + Sync
+impl<'a, A: KVStore + ?Sized, CM: Deref, L: Deref, S: WriteableScore<'a>> Persister<'a, CM, L, S> for A
 where
        CM::Target: 'static + AChannelManager,
        L::Target: 'static + Logger,
@@ -212,7 +202,7 @@ where
        }
 }
 
-impl<ChannelSigner: WriteableEcdsaChannelSigner, K: KVStore> Persist<ChannelSigner> for K {
+impl<ChannelSigner: WriteableEcdsaChannelSigner, K: KVStore + ?Sized> Persist<ChannelSigner> for K {
        // TODO: We really need a way for the persister to inform the user that its time to crash/shut
        // down once these start returning failure.
        // Then we should return InProgress rather than UnrecoverableError, implying we should probably
@@ -241,36 +231,32 @@ impl<ChannelSigner: WriteableEcdsaChannelSigner, K: KVStore> Persist<ChannelSign
                        Err(_) => chain::ChannelMonitorUpdateStatus::UnrecoverableError
                }
        }
-}
-
-impl<ChannelSigner: WriteableEcdsaChannelSigner> Persist<ChannelSigner> for dyn KVStore + Send + Sync {
-       // TODO: We really need a way for the persister to inform the user that its time to crash/shut
-       // down once these start returning failure.
-       // Then we should return InProgress rather than UnrecoverableError, implying we should probably
-       // just shut down the node since we're not retrying persistence!
 
-       fn persist_new_channel(&self, funding_txo: OutPoint, monitor: &ChannelMonitor<ChannelSigner>, _update_id: MonitorUpdateId) -> chain::ChannelMonitorUpdateStatus {
-               let key = format!("{}_{}", funding_txo.txid.to_string(), funding_txo.index);
-               match self.write(
+       fn archive_persisted_channel(&self, funding_txo: OutPoint) {
+               let monitor_name = MonitorName::from(funding_txo);
+               let monitor = match self.read(
                        CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
                        CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
-                       &key, &monitor.encode())
-               {
-                       Ok(()) => chain::ChannelMonitorUpdateStatus::Completed,
-                       Err(_) => chain::ChannelMonitorUpdateStatus::UnrecoverableError
-               }
-       }
-
-       fn update_persisted_channel(&self, funding_txo: OutPoint, _update: Option<&ChannelMonitorUpdate>, monitor: &ChannelMonitor<ChannelSigner>, _update_id: MonitorUpdateId) -> chain::ChannelMonitorUpdateStatus {
-               let key = format!("{}_{}", funding_txo.txid.to_string(), funding_txo.index);
+                       monitor_name.as_str(),
+               ) {
+                       Ok(monitor) => monitor,
+                       Err(_) => return
+               };
                match self.write(
+                       ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
+                       ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
+                       monitor_name.as_str(),
+                       &monitor,
+               ) {
+                       Ok(()) => {}
+                       Err(_e) => return
+               };
+               let _ = self.remove(
                        CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
                        CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
-                       &key, &monitor.encode())
-               {
-                       Ok(()) => chain::ChannelMonitorUpdateStatus::Completed,
-                       Err(_) => chain::ChannelMonitorUpdateStatus::UnrecoverableError
-               }
+                       monitor_name.as_str(),
+                       true,
+               );
        }
 }
 
@@ -778,6 +764,29 @@ where
                        self.persist_new_channel(funding_txo, monitor, monitor_update_call_id)
                }
        }
+
+       fn archive_persisted_channel(&self, funding_txo: OutPoint) {
+               let monitor_name = MonitorName::from(funding_txo);
+               let monitor = match self.read_monitor(&monitor_name) {
+                       Ok((_block_hash, monitor)) => monitor,
+                       Err(_) => return
+               };
+               match self.kv_store.write(
+                       ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
+                       ARCHIVED_CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
+                       monitor_name.as_str(),
+                       &monitor.encode()
+               ) {
+                       Ok(()) => {},
+                       Err(_e) => return,
+               };
+               let _ = self.kv_store.remove(
+                       CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
+                       CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
+                       monitor_name.as_str(),
+                       true,
+               );
+       }
 }
 
 impl<K: Deref, L: Deref, ES: Deref, SP: Deref> MonitorUpdatingPersister<K, L, ES, SP>
@@ -895,12 +904,13 @@ impl From<u64> for UpdateName {
 #[cfg(test)]
 mod tests {
        use super::*;
-       use crate::chain::chainmonitor::Persist;
        use crate::chain::ChannelMonitorUpdateStatus;
        use crate::events::{ClosureReason, MessageSendEventsProvider};
        use crate::ln::functional_test_utils::*;
        use crate::util::test_utils::{self, TestLogger, TestStore};
        use crate::{check_added_monitors, check_closed_broadcast};
+       use crate::sync::Arc;
+       use crate::util::test_channel_signer::TestChannelSigner;
 
        const EXPECTED_UPDATES_PER_PAYMENT: u64 = 5;
 
@@ -1241,4 +1251,14 @@ mod tests {
                        .read(CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE, monitor_name.as_str(), UpdateName::from(u64::MAX - 1).as_str())
                        .is_err());
        }
+
+       fn persist_fn<P: Deref, ChannelSigner: WriteableEcdsaChannelSigner>(_persist: P) -> bool where P::Target: Persist<ChannelSigner> {
+               true
+       }
+
+       #[test]
+       fn kvstore_trait_object_usage() {
+               let store: Arc<dyn KVStore + Send + Sync> = Arc::new(TestStore::new(false));
+               assert!(persist_fn::<_, TestChannelSigner>(store.clone()));
+       }
 }