Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
121 changes: 59 additions & 62 deletions lightning-background-processor/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -53,11 +53,10 @@ use lightning::sign::{
};
use lightning::util::logger::Logger;
use lightning::util::persist::{
KVStore, KVStoreSync, KVStoreSyncWrapper, CHANNEL_MANAGER_PERSISTENCE_KEY,
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_KEY, NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, SCORER_PERSISTENCE_KEY,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE, SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
KVStore, KVStoreSync, KVStoreSyncWrapper, CHANNEL_MANAGER_KEY,
CHANNEL_MANAGER_PRIMARY_NAMESPACE, CHANNEL_MANAGER_SECONDARY_NAMESPACE, NETWORK_GRAPH_KEY,
NETWORK_GRAPH_PRIMARY_NAMESPACE, NETWORK_GRAPH_SECONDARY_NAMESPACE, SCORER_KEY,
SCORER_PRIMARY_NAMESPACE, SCORER_SECONDARY_NAMESPACE,
};
use lightning::util::sweep::{OutputSweeper, OutputSweeperSync};
#[cfg(feature = "std")]
Expand Down Expand Up @@ -944,9 +943,9 @@ where
log_trace!(logger, "Persisting scorer after update");
if let Err(e) = kv_store
.write(
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
SCORER_PERSISTENCE_KEY,
SCORER_PRIMARY_NAMESPACE,
SCORER_SECONDARY_NAMESPACE,
SCORER_KEY,
scorer.encode(),
)
.await
Expand Down Expand Up @@ -1080,9 +1079,9 @@ where
let fut = async {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here and below, we could do:

diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs
index 7b4f29896..af851f5f6 100644
--- a/lightning-background-processor/src/lib.rs
+++ b/lightning-background-processor/src/lib.rs
@@ -1068,15 +1068,9 @@ where
                if channel_manager.get_cm().get_and_clear_needs_persistence() {
                        log_trace!(logger, "Persisting ChannelManager...");

+                       let mgr = channel_manager.get_cm().encode();
                        let fut = async {
-                               kv_store
-                                       .write(
-                                               CHANNEL_MANAGER_NAMESPACE,
-                                               "",
-                                               CHANNEL_MANAGER_KEY,
-                                               channel_manager.get_cm().encode(),
-                                       )
-                                       .await
+                               kv_store.write(CHANNEL_MANAGER_NAMESPACE, "", CHANNEL_MANAGER_KEY, mgr).await
                        };
                        // TODO: Once our MSRV is 1.68 we should be able to drop the Box
                        let mut fut = Box::pin(fut);

kv_store
.write(
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
CHANNEL_MANAGER_PERSISTENCE_KEY,
CHANNEL_MANAGER_PRIMARY_NAMESPACE,
CHANNEL_MANAGER_SECONDARY_NAMESPACE,
CHANNEL_MANAGER_KEY,
channel_manager.get_cm().encode(),
)
.await
Expand Down Expand Up @@ -1143,9 +1142,9 @@ where
let fut = async {
if let Err(e) = kv_store
.write(
NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_KEY,
NETWORK_GRAPH_PRIMARY_NAMESPACE,
NETWORK_GRAPH_SECONDARY_NAMESPACE,
NETWORK_GRAPH_KEY,
network_graph.encode(),
)
.await
Expand Down Expand Up @@ -1185,9 +1184,9 @@ where
let fut = async {
if let Err(e) = kv_store
.write(
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
SCORER_PERSISTENCE_KEY,
SCORER_PRIMARY_NAMESPACE,
SCORER_SECONDARY_NAMESPACE,
SCORER_KEY,
scorer.encode(),
)
.await
Expand Down Expand Up @@ -1301,28 +1300,28 @@ where
// ChannelMonitor update(s) persisted without a corresponding ChannelManager update.
kv_store
.write(
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
CHANNEL_MANAGER_PERSISTENCE_KEY,
CHANNEL_MANAGER_PRIMARY_NAMESPACE,
CHANNEL_MANAGER_SECONDARY_NAMESPACE,
CHANNEL_MANAGER_KEY,
channel_manager.get_cm().encode(),
)
.await?;
if let Some(ref scorer) = scorer {
kv_store
.write(
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
SCORER_PERSISTENCE_KEY,
SCORER_PRIMARY_NAMESPACE,
SCORER_SECONDARY_NAMESPACE,
SCORER_KEY,
scorer.encode(),
)
.await?;
}
if let Some(network_graph) = gossip_sync.network_graph() {
kv_store
.write(
NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_KEY,
NETWORK_GRAPH_PRIMARY_NAMESPACE,
NETWORK_GRAPH_SECONDARY_NAMESPACE,
NETWORK_GRAPH_KEY,
network_graph.encode(),
)
.await?;
Expand Down Expand Up @@ -1527,9 +1526,9 @@ impl BackgroundProcessor {
if update_scorer(scorer, &event, duration_since_epoch) {
log_trace!(logger, "Persisting scorer after update");
if let Err(e) = kv_store.write(
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
SCORER_PERSISTENCE_KEY,
SCORER_PRIMARY_NAMESPACE,
SCORER_SECONDARY_NAMESPACE,
SCORER_KEY,
scorer.encode(),
) {
log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e)
Expand Down Expand Up @@ -1627,9 +1626,9 @@ impl BackgroundProcessor {
if channel_manager.get_cm().get_and_clear_needs_persistence() {
log_trace!(logger, "Persisting ChannelManager...");
(kv_store.write(
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
CHANNEL_MANAGER_PERSISTENCE_KEY,
CHANNEL_MANAGER_PRIMARY_NAMESPACE,
CHANNEL_MANAGER_SECONDARY_NAMESPACE,
CHANNEL_MANAGER_KEY,
channel_manager.get_cm().encode(),
))?;
log_trace!(logger, "Done persisting ChannelManager.");
Expand Down Expand Up @@ -1666,9 +1665,9 @@ impl BackgroundProcessor {
duration_since_epoch.as_secs(),
);
if let Err(e) = kv_store.write(
NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_KEY,
NETWORK_GRAPH_PRIMARY_NAMESPACE,
NETWORK_GRAPH_SECONDARY_NAMESPACE,
NETWORK_GRAPH_KEY,
network_graph.encode(),
) {
log_error!(logger, "Error: Failed to persist network graph, check your disk and permissions {}", e);
Expand All @@ -1695,9 +1694,9 @@ impl BackgroundProcessor {
log_trace!(logger, "Calling time_passed and persisting scorer");
scorer.write_lock().time_passed(duration_since_epoch);
if let Err(e) = kv_store.write(
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
SCORER_PERSISTENCE_KEY,
SCORER_PRIMARY_NAMESPACE,
SCORER_SECONDARY_NAMESPACE,
SCORER_KEY,
scorer.encode(),
) {
log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e);
Expand Down Expand Up @@ -1735,24 +1734,24 @@ impl BackgroundProcessor {
// some races where users quit while channel updates were in-flight, with
// ChannelMonitor update(s) persisted without a corresponding ChannelManager update.
kv_store.write(
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
CHANNEL_MANAGER_PERSISTENCE_KEY,
CHANNEL_MANAGER_PRIMARY_NAMESPACE,
CHANNEL_MANAGER_SECONDARY_NAMESPACE,
CHANNEL_MANAGER_KEY,
channel_manager.get_cm().encode(),
)?;
if let Some(ref scorer) = scorer {
kv_store.write(
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
SCORER_PERSISTENCE_KEY,
SCORER_PRIMARY_NAMESPACE,
SCORER_SECONDARY_NAMESPACE,
SCORER_KEY,
scorer.encode(),
)?;
}
if let Some(network_graph) = gossip_sync.network_graph() {
kv_store.write(
NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_KEY,
NETWORK_GRAPH_PRIMARY_NAMESPACE,
NETWORK_GRAPH_SECONDARY_NAMESPACE,
NETWORK_GRAPH_KEY,
network_graph.encode(),
)?;
}
Expand Down Expand Up @@ -1844,12 +1843,10 @@ mod tests {
use lightning::types::payment::PaymentHash;
use lightning::util::config::UserConfig;
use lightning::util::persist::{
KVStoreSync, KVStoreSyncWrapper, CHANNEL_MANAGER_PERSISTENCE_KEY,
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_KEY,
NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE,
SCORER_PERSISTENCE_KEY, SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
KVStoreSync, KVStoreSyncWrapper, CHANNEL_MANAGER_KEY, CHANNEL_MANAGER_PRIMARY_NAMESPACE,
CHANNEL_MANAGER_SECONDARY_NAMESPACE, NETWORK_GRAPH_KEY, NETWORK_GRAPH_PRIMARY_NAMESPACE,
NETWORK_GRAPH_SECONDARY_NAMESPACE, SCORER_KEY, SCORER_PRIMARY_NAMESPACE,
SCORER_SECONDARY_NAMESPACE,
};
use lightning::util::ser::Writeable;
use lightning::util::sweep::{
Expand Down Expand Up @@ -2104,18 +2101,18 @@ mod tests {
fn write(
&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
) -> lightning::io::Result<()> {
if primary_namespace == CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE
&& secondary_namespace == CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE
&& key == CHANNEL_MANAGER_PERSISTENCE_KEY
if primary_namespace == CHANNEL_MANAGER_PRIMARY_NAMESPACE
&& secondary_namespace == CHANNEL_MANAGER_SECONDARY_NAMESPACE
&& key == CHANNEL_MANAGER_KEY
{
if let Some((error, message)) = self.manager_error {
return Err(std::io::Error::new(error, message).into());
}
}

if primary_namespace == NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE
&& secondary_namespace == NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE
&& key == NETWORK_GRAPH_PERSISTENCE_KEY
if primary_namespace == NETWORK_GRAPH_PRIMARY_NAMESPACE
&& secondary_namespace == NETWORK_GRAPH_SECONDARY_NAMESPACE
&& key == NETWORK_GRAPH_KEY
{
if let Some(sender) = &self.graph_persistence_notifier {
match sender.send(()) {
Expand All @@ -2131,9 +2128,9 @@ mod tests {
}
}

if primary_namespace == SCORER_PERSISTENCE_PRIMARY_NAMESPACE
&& secondary_namespace == SCORER_PERSISTENCE_SECONDARY_NAMESPACE
&& key == SCORER_PERSISTENCE_KEY
if primary_namespace == SCORER_PRIMARY_NAMESPACE
&& secondary_namespace == SCORER_SECONDARY_NAMESPACE
&& key == SCORER_KEY
{
if let Some((error, message)) = self.scorer_error {
return Err(std::io::Error::new(error, message).into());
Expand Down
65 changes: 16 additions & 49 deletions lightning/src/ln/chanmon_update_fail_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -28,9 +28,8 @@ use crate::ln::types::ChannelId;
use crate::sign::NodeSigner;
use crate::util::native_async::FutureQueue;
use crate::util::persist::{
MonitorName, MonitorUpdatingPersisterAsync, CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
MonitorName, MonitorUpdatingPersisterAsync, CHANNEL_MONITOR_PRIMARY_NAMESPACE,
CHANNEL_MONITOR_SECONDARY_NAMESPACE, MONITOR_UPDATE_PRIMARY_NAMESPACE,
};
use crate::util::ser::{ReadableArgs, Writeable};
use crate::util::test_channel_signer::TestChannelSigner;
Expand Down Expand Up @@ -4940,8 +4939,8 @@ fn native_async_persist() {
let funding_txo = OutPoint { txid: funding_tx.compute_txid(), index: 0 };
let key = MonitorName::V1Channel(funding_txo).to_string();
let pending_writes = kv_store.list_pending_async_writes(
CHANNEL_MONITOR_PERSISTENCE_PRIMARY_NAMESPACE,
CHANNEL_MONITOR_PERSISTENCE_SECONDARY_NAMESPACE,
CHANNEL_MONITOR_PRIMARY_NAMESPACE,
CHANNEL_MONITOR_SECONDARY_NAMESPACE,
&key,
);
assert_eq!(pending_writes.len(), 1);
Expand Down Expand Up @@ -4971,37 +4970,21 @@ fn native_async_persist() {
persist_futures.poll_futures();
assert_eq!(async_chain_monitor.release_pending_monitor_events().len(), 0);

let pending_writes = kv_store.list_pending_async_writes(
CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
&key,
"1",
);
let pending_writes =
kv_store.list_pending_async_writes(MONITOR_UPDATE_PRIMARY_NAMESPACE, &key, "1");
assert_eq!(pending_writes.len(), 1);
let pending_writes = kv_store.list_pending_async_writes(
CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
&key,
"2",
);
let pending_writes =
kv_store.list_pending_async_writes(MONITOR_UPDATE_PRIMARY_NAMESPACE, &key, "2");
assert_eq!(pending_writes.len(), 1);

kv_store.complete_async_writes_through(
CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
&key,
"1",
usize::MAX,
);
kv_store.complete_async_writes_through(MONITOR_UPDATE_PRIMARY_NAMESPACE, &key, "1", usize::MAX);
persist_futures.poll_futures();
// While the `ChainMonitor` could return a `MonitorEvent::Completed` here, it currently
// doesn't. If that ever changes we should validate that the `Completed` event has the correct
// `monitor_update_id` (1).
assert!(async_chain_monitor.release_pending_monitor_events().is_empty());

kv_store.complete_async_writes_through(
CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
&key,
"2",
usize::MAX,
);
kv_store.complete_async_writes_through(MONITOR_UPDATE_PRIMARY_NAMESPACE, &key, "2", usize::MAX);
persist_futures.poll_futures();
let completed_persist = async_chain_monitor.release_pending_monitor_events();
assert_eq!(completed_persist.len(), 1);
Expand All @@ -5020,34 +5003,18 @@ fn native_async_persist() {
persist_futures.poll_futures();
assert_eq!(async_chain_monitor.release_pending_monitor_events().len(), 0);

let pending_writes = kv_store.list_pending_async_writes(
CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
&key,
"3",
);
let pending_writes =
kv_store.list_pending_async_writes(MONITOR_UPDATE_PRIMARY_NAMESPACE, &key, "3");
assert_eq!(pending_writes.len(), 1);
let pending_writes = kv_store.list_pending_async_writes(
CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
&key,
"4",
);
let pending_writes =
kv_store.list_pending_async_writes(MONITOR_UPDATE_PRIMARY_NAMESPACE, &key, "4");
assert_eq!(pending_writes.len(), 1);

kv_store.complete_async_writes_through(
CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
&key,
"4",
usize::MAX,
);
kv_store.complete_async_writes_through(MONITOR_UPDATE_PRIMARY_NAMESPACE, &key, "4", usize::MAX);
persist_futures.poll_futures();
assert_eq!(async_chain_monitor.release_pending_monitor_events().len(), 0);

kv_store.complete_async_writes_through(
CHANNEL_MONITOR_UPDATE_PERSISTENCE_PRIMARY_NAMESPACE,
&key,
"3",
usize::MAX,
);
kv_store.complete_async_writes_through(MONITOR_UPDATE_PRIMARY_NAMESPACE, &key, "3", usize::MAX);
persist_futures.poll_futures();
let completed_persist = async_chain_monitor.release_pending_monitor_events();
assert_eq!(completed_persist.len(), 1);
Expand Down
Loading