Skip to content

Commit

Permalink
Add ChainMonitor::archive_fully_resolved_monitor_channels
Browse files Browse the repository at this point in the history
  Archives fully resolved channel monitors by adding them to a backup
  location and removing them from the primary storage & the monitor set.

  This is useful for pruning fully resolved monitors from the monitor
  set and primary storage so they are not reloaded on every new new
  block connection.

  We also add a new function, `archive_persisted_channel` to the
  `Persist` trait that writes the monitor to an archive storage and
  removes it from the primary storage.
  • Loading branch information
jbesraa committed Apr 9, 2024
1 parent 16a1dfb commit 664a2ed
Show file tree
Hide file tree
Showing 5 changed files with 123 additions and 1 deletion.
3 changes: 3 additions & 0 deletions fuzz/src/utils/test_persister.rs
Original file line number Diff line number Diff line change
Expand Up @@ -17,4 +17,7 @@ impl chainmonitor::Persist<TestChannelSigner> for TestPersister {
fn update_persisted_channel(&self, _funding_txo: OutPoint, _update: Option<&channelmonitor::ChannelMonitorUpdate>, _data: &channelmonitor::ChannelMonitor<TestChannelSigner>, _update_id: MonitorUpdateId) -> chain::ChannelMonitorUpdateStatus {
self.update_ret.lock().unwrap().clone()
}

fn archive_persisted_channel(&self, _: OutPoint) {
}
}
26 changes: 26 additions & 0 deletions lightning/src/chain/chainmonitor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -194,6 +194,11 @@ pub trait Persist<ChannelSigner: WriteableEcdsaChannelSigner> {
///
/// [`Writeable::write`]: crate::util::ser::Writeable::write
fn update_persisted_channel(&self, channel_funding_outpoint: OutPoint, update: Option<&ChannelMonitorUpdate>, data: &ChannelMonitor<ChannelSigner>, update_id: MonitorUpdateId) -> ChannelMonitorUpdateStatus;
/// Prevents the channel monitor from being loaded on startup.
///
/// Archiving the data in a backup location (rather than deleting it fully) is useful for
/// hedging against data loss in case
fn archive_persisted_channel(&self, channel_funding_outpoint: OutPoint);
}

struct MonitorHolder<ChannelSigner: WriteableEcdsaChannelSigner> {
Expand Down Expand Up @@ -656,6 +661,27 @@ where C::Target: chain::Filter,
}
}
}

/// Archives fully resolved channel monitors by calling [`Persist::archive_persisted_channel`].
///
/// This is useful for pruning fully resolved monitors from the monitor set and primary
/// storage so they are not reloaded on every new new block connection.
///
/// Depending on the `[Persist::archive_persisted_channel]` the monitor data could be moved
/// to an archive location or/and removed entirely.
pub fn archive_fully_resolved_channel_monitors(&self, to_archive: Vec<OutPoint>) {
let mut monitors = self.monitors.write().unwrap();
for funding_txo in to_archive {
let channel_monitor = monitors.get(&funding_txo);
if let Some(channel_monitor) = channel_monitor {
if channel_monitor.monitor.is_fully_resolved() {
self.persister.archive_persisted_channel(funding_txo);
monitors.remove(&funding_txo);
};
};
}
}

}

impl<ChannelSigner: WriteableEcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref>
Expand Down
17 changes: 16 additions & 1 deletion lightning/src/ln/monitor_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -171,7 +171,7 @@ fn do_chanmon_claim_value_coop_close(anchors: bool) {
user_config.manually_accept_inbound_channels = true;
}
let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[Some(user_config), Some(user_config)]);
let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs);

let (_, _, chan_id, funding_tx) =
create_announced_chan_between_nodes_with_value(&nodes, 0, 1, 1_000_000, 1_000_000);
Expand Down Expand Up @@ -259,6 +259,21 @@ fn do_chanmon_claim_value_coop_close(anchors: bool) {

check_closed_event!(nodes[0], 1, ClosureReason::LocallyInitiatedCooperativeClosure, [nodes[1].node.get_our_node_id()], 1000000);
check_closed_event!(nodes[1], 1, ClosureReason::CounterpartyInitiatedCooperativeClosure, [nodes[0].node.get_our_node_id()], 1000000);

assert!(nodes[0].chain_monitor.chain_monitor.get_and_clear_pending_events().is_empty());
assert!(nodes[1].chain_monitor.chain_monitor.get_and_clear_pending_events().is_empty());

// Test we can archived fully resolved channel monitor.
assert_eq!(nodes[0].chain_monitor.chain_monitor.list_monitors().len(), 1); // one monitor
// first archive should set balances_empty_height to current block height
nodes[0].chain_monitor.chain_monitor.archive_fully_resolved_channel_monitors(vec![funding_outpoint]);
connect_blocks(&nodes[0], 2016);
// Second call after 2016 blocks, should archive the monitor
nodes[0].chain_monitor.chain_monitor.archive_fully_resolved_channel_monitors(vec![funding_outpoint]);
assert_eq!(nodes[0].chain_monitor.chain_monitor.list_monitors().len(), 0); // no monitor
// cleanup
nodes.get_mut(0).unwrap().chain_source.remove_watched_txn_and_outputs(funding_outpoint,
funding_tx.txid(), funding_tx.output[0].script_pubkey.clone());
}

#[test]
Expand Down
58 changes: 58 additions & 0 deletions lightning/src/util/persist.rs
Original file line number Diff line number Diff line change
Expand Up @@ -56,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.
Expand Down Expand Up @@ -212,6 +217,34 @@ impl<ChannelSigner: WriteableEcdsaChannelSigner, K: KVStore + ?Sized> Persist<Ch
Err(_) => chain::ChannelMonitorUpdateStatus::UnrecoverableError
}
}

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,
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 key = format!("{}_{}", funding_txo.txid.to_string(), funding_txo.index);
let _ = self.remove(
CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
&key,
false
);
}
}

/// Read previously persisted [`ChannelMonitor`]s from the store.
Expand Down Expand Up @@ -718,6 +751,31 @@ 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(),
false,
);
}
}

impl<K: Deref, L: Deref, ES: Deref, SP: Deref> MonitorUpdatingPersister<K, L, ES, SP>
Expand Down
20 changes: 20 additions & 0 deletions lightning/src/util/test_utils.rs
Original file line number Diff line number Diff line change
Expand Up @@ -504,6 +504,10 @@ impl<Signer: sign::ecdsa::WriteableEcdsaChannelSigner> chainmonitor::Persist<Sig
}
res
}

fn archive_persisted_channel(&self, funding_txo: OutPoint) {
<TestPersister as chainmonitor::Persist<TestChannelSigner>>::archive_persisted_channel(&self.persister, funding_txo);
}
}

pub struct TestPersister {
Expand Down Expand Up @@ -552,6 +556,18 @@ impl<Signer: sign::ecdsa::WriteableEcdsaChannelSigner> chainmonitor::Persist<Sig
}
ret
}

fn archive_persisted_channel(&self, funding_txo: OutPoint) {
// remove the channel from the offchain_monitor_updates map
match self.offchain_monitor_updates.lock().unwrap().remove(&funding_txo) {
Some(_) => {},
None => {
// If the channel was not in the offchain_monitor_updates map, it should be in the
// chain_sync_monitor_persistences map.
assert!(self.chain_sync_monitor_persistences.lock().unwrap().remove(&funding_txo).is_some());
}
};
}
}

pub struct TestStore {
Expand Down Expand Up @@ -1363,6 +1379,10 @@ impl TestChainSource {
watched_outputs: Mutex::new(new_hash_set()),
}
}
pub fn remove_watched_txn_and_outputs(&self, outpoint: OutPoint, txid: Txid, script_pubkey: ScriptBuf) {
self.watched_outputs.lock().unwrap().remove(&(outpoint, script_pubkey.clone()));
self.watched_txn.lock().unwrap().remove(&(txid, script_pubkey));
}
}

impl UtxoLookup for TestChainSource {
Expand Down

0 comments on commit 664a2ed

Please sign in to comment.