Skip to content

Commit

Permalink
Stop cleaning monitor updates on new block connect
Browse files Browse the repository at this point in the history 
Previously, we used to cleanup monitor updates at both consolidation
threshold and new block connects. With this change we will only
cleanup when our consolidation criteria is met. Also, we remove
monitor read from cleanup logic, in case of update consolidation.
Note: In case of channel-closing monitor update, we still need to
read the old monitor before persisting the new one in order to
determine the cleanup range.
  • Loading branch information
@G8XSU
G8XSU committed Dec 14, 2023
1 parent c2bbfff commit ef09096
Showing 1 changed file with 82 additions and 89 deletions.
171 changes: 82 additions & 89 deletions lightning/src/util/persist.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -346,9 +346,10 @@ where
///
/// # Pruning stale channel updates
///
/// Stale updates are pruned when a full monitor is written. The old monitor is first read, and if
/// that succeeds, updates in the range between the old and new monitors are deleted. The `lazy`
/// flag is used on the [`KVStore::remove`] method, so there are no guarantees that the deletions
/// Stale updates are pruned when the consolidation threshold is reached according to `maximum_pending_updates`.
/// Monitor updates in the range between the latest `update_id` and `update_id - maximum_pending_updates`
/// are deleted.
/// The `lazy` flag is used on the [`KVStore::remove`] method, so there are no guarantees that the deletions
/// will complete. However, stale updates are not a problem for data integrity, since updates are
/// only read that are higher than the stored [`ChannelMonitor`]'s `update_id`.
///
Expand Down Expand Up @@ -610,24 +611,6 @@ where
) -> chain::ChannelMonitorUpdateStatus {
// Determine the proper key for this monitor
let monitor_name = MonitorName::from(funding_txo);
let maybe_old_monitor = self.read_monitor(&monitor_name);
match maybe_old_monitor {
Ok((_, ref old_monitor)) => {
// Check that this key isn't already storing a monitor with a higher update_id
// (collision)
if old_monitor.get_latest_update_id() > monitor.get_latest_update_id() {
log_error!(
self.logger,
"Tried to write a monitor at the same outpoint {} with a higher update_id!",
monitor_name.as_str()
);
return chain::ChannelMonitorUpdateStatus::UnrecoverableError;
}
}
// This means the channel monitor is new.
Err(ref e) if e.kind() == io::ErrorKind::NotFound => {}
_ => return chain::ChannelMonitorUpdateStatus::UnrecoverableError,
}
// Serialize and write the new monitor
let mut monitor_bytes = Vec::with_capacity(
MONITOR_UPDATING_PERSISTER_PREPEND_SENTINEL.len() + monitor.serialized_length(),
Expand All @@ -641,65 +624,12 @@ where
&monitor_bytes,
) {
Ok(_) => {
// Assess cleanup. Typically, we'll clean up only between the last two known full
// monitors.
if let Ok((_, old_monitor)) = maybe_old_monitor {
let start = old_monitor.get_latest_update_id();
let end = if monitor.get_latest_update_id() == CLOSED_CHANNEL_UPDATE_ID {
// We don't want to clean the rest of u64, so just do possible pending
// updates. Note that we never write updates at
// `CLOSED_CHANNEL_UPDATE_ID`.
cmp::min(
start.saturating_add(self.maximum_pending_updates),
CLOSED_CHANNEL_UPDATE_ID - 1,
)
} else {
monitor.get_latest_update_id().saturating_sub(1)
};
// We should bother cleaning up only if there's at least one update
// expected.
for update_id in start..=end {
let update_name = UpdateName::from(update_id);
#[cfg(debug_assertions)]
{
if let Ok(update) =
self.read_monitor_update(&monitor_name, &update_name)
{
// Assert that we are reading what we think we are.
debug_assert_eq!(update.update_id, update_name.0);
} else if update_id != start && monitor.get_latest_update_id() != CLOSED_CHANNEL_UPDATE_ID
{
// We're deleting something we should know doesn't exist.
panic!(
"failed to read monitor update {}",
update_name.as_str()
);
}
// On closed channels, we will unavoidably try to read
// non-existent updates since we have to guess at the range of
// stale updates, so do nothing.
}
if let Err(e) = self.kv_store.remove(
CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
monitor_name.as_str(),
update_name.as_str(),
true,
) {
log_error!(
self.logger,
"error cleaning up channel monitor updates for monitor {}, reason: {}",
monitor_name.as_str(),
e
);
};
}
};
chain::ChannelMonitorUpdateStatus::Completed
}
Err(e) => {
log_error!(
self.logger,
"error writing channel monitor {}/{}/{} reason: {}",
"Failed to write ChannelMonitor {}/{}/{} reason: {}",
CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
monitor_name.as_str(),
Expand Down Expand Up @@ -741,7 +671,7 @@ where
Err(e) => {
log_error!(
self.logger,
"error writing channel monitor update {}/{}/{} reason: {}",
"Failed to write ChannelMonitorUpdate {}/{}/{} reason: {}",
CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
monitor_name.as_str(),
update_name.as_str(),
Expand All @@ -751,8 +681,41 @@ where
}
}
} else {
// We could write this update, but it meets criteria of our design that call for a full monitor write.
self.persist_new_channel(funding_txo, monitor, monitor_update_call_id)
let monitor_name = MonitorName::from(funding_txo);
// In case of channel-close monitor update, we need to read old monitor before persisting
// the new one in order to determine the cleanup range.
let maybe_old_monitor = match monitor.get_latest_update_id() {
CLOSED_CHANNEL_UPDATE_ID => self.read_monitor(&monitor_name).ok(),
_ => None
};

// We could write this update, but it meets criteria of our design that calls for a full monitor write.
let monitor_update_status = self.persist_new_channel(funding_txo, monitor, monitor_update_call_id);

if let chain::ChannelMonitorUpdateStatus::Completed = monitor_update_status {
let cleanup_range = if monitor.get_latest_update_id() == CLOSED_CHANNEL_UPDATE_ID {
// If there is an error while reading old monitor, we skip clean up.
maybe_old_monitor.map(|(_, ref old_monitor)| {
let start = old_monitor.get_latest_update_id();
// We never persist an update with update_id = CLOSED_CHANNEL_UPDATE_ID
let end = cmp::min(
start.saturating_add(self.maximum_pending_updates),
CLOSED_CHANNEL_UPDATE_ID - 1,
);
(start, end)
})
} else {
let end = monitor.get_latest_update_id();
let start = end.saturating_sub(self.maximum_pending_updates);
Some((start, end))
};

if let Some((start, end)) = cleanup_range {
self.cleanup_in_range(monitor_name, start, end);
}
}

monitor_update_status
}
} else {
// There is no update given, so we must persist a new monitor.
Expand All @@ -761,6 +724,34 @@ where
}
}

impl<K: Deref, L: Deref, ES: Deref, SP: Deref> MonitorUpdatingPersister<K, L, ES, SP>
where
ES::Target: EntropySource + Sized,
K::Target: KVStore,
L::Target: Logger,
SP::Target: SignerProvider + Sized
{
// Cleans up monitor updates for given monitor in range `start..=end`.
fn cleanup_in_range(&self, monitor_name: MonitorName, start: u64, end: u64) {
for update_id in start..=end {
let update_name = UpdateName::from(update_id);
if let Err(e) = self.kv_store.remove(
CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
monitor_name.as_str(),
update_name.as_str(),
true,
) {
log_error!(
self.logger,
"Failed to clean up channel monitor updates for monitor {}, reason: {}",
monitor_name.as_str(),
e
);
};
}
}
}

/// A struct representing a name for a monitor.
#[derive(Debug)]
struct MonitorName(String);
Expand Down Expand Up @@ -896,20 +887,21 @@ mod tests {
#[test]
fn persister_with_real_monitors() {
// This value is used later to limit how many iterations we perform.
let test_max_pending_updates = 7;
let persister_0_max_pending_updates = 7;
// Intentionally set this to a smaller value to test a different alignment.
let persister_1_max_pending_updates = 3;
let chanmon_cfgs = create_chanmon_cfgs(4);
let persister_0 = MonitorUpdatingPersister {
kv_store: &TestStore::new(false),
logger: &TestLogger::new(),
maximum_pending_updates: test_max_pending_updates,
maximum_pending_updates: persister_0_max_pending_updates,
entropy_source: &chanmon_cfgs[0].keys_manager,
signer_provider: &chanmon_cfgs[0].keys_manager,
};
let persister_1 = MonitorUpdatingPersister {
kv_store: &TestStore::new(false),
logger: &TestLogger::new(),
// Intentionally set this to a smaller value to test a different alignment.
maximum_pending_updates: 3,
maximum_pending_updates: persister_1_max_pending_updates,
entropy_source: &chanmon_cfgs[1].keys_manager,
signer_provider: &chanmon_cfgs[1].keys_manager,
};
Expand All @@ -934,7 +926,6 @@ mod tests {
node_cfgs[1].chain_monitor = chain_mon_1;
let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
let nodes = create_network(2, &node_cfgs, &node_chanmgrs);

let broadcaster_0 = &chanmon_cfgs[2].tx_broadcaster;
let broadcaster_1 = &chanmon_cfgs[3].tx_broadcaster;

Expand All @@ -957,10 +948,11 @@ mod tests {
for (_, mon) in persisted_chan_data_0.iter() {
// check that when we read it, we got the right update id
assert_eq!(mon.get_latest_update_id(), $expected_update_id);
// if the CM is at the correct update id without updates, ensure no updates are stored

// if the CM is at consolidation threshold, ensure no updates are stored.
let monitor_name = MonitorName::from(mon.get_funding_txo().0);
let (_, cm_0) = persister_0.read_monitor(&monitor_name).unwrap();
if cm_0.get_latest_update_id() == $expected_update_id {
if mon.get_latest_update_id() % persister_0_max_pending_updates == 0
|| mon.get_latest_update_id() == CLOSED_CHANNEL_UPDATE_ID {
assert_eq!(
persister_0.kv_store.list(CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
monitor_name.as_str()).unwrap().len(),
Expand All @@ -975,8 +967,9 @@ mod tests {
for (_, mon) in persisted_chan_data_1.iter() {
assert_eq!(mon.get_latest_update_id(), $expected_update_id);
let monitor_name = MonitorName::from(mon.get_funding_txo().0);
let (_, cm_1) = persister_1.read_monitor(&monitor_name).unwrap();
if cm_1.get_latest_update_id() == $expected_update_id {
// if the CM is at consolidation threshold, ensure no updates are stored.
if mon.get_latest_update_id() % persister_1_max_pending_updates == 0
|| mon.get_latest_update_id() == CLOSED_CHANNEL_UPDATE_ID {
assert_eq!(
persister_1.kv_store.list(CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
monitor_name.as_str()).unwrap().len(),
Expand All @@ -1001,7 +994,7 @@ mod tests {
// Send a few more payments to try all the alignments of max pending updates with
// updates for a payment sent and received.
let mut sender = 0;
for i in 3..=test_max_pending_updates * 2 {
for i in 3..=persister_0_max_pending_updates * 2 {
let receiver;
if sender == 0 {
sender = 1;
Expand Down

0 comments on commit ef09096

Please sign in to comment.