Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
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
136 changes: 44 additions & 92 deletions lightning-background-processor/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -53,11 +53,8 @@ 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_NAMESPACE,
NETWORK_GRAPH_KEY, NETWORK_GRAPH_NAMESPACE, SCORER_KEY, SCORER_NAMESPACE,
};
use lightning::util::sweep::{OutputSweeper, OutputSweeperSync};
#[cfg(feature = "std")]
Expand Down Expand Up @@ -942,14 +939,8 @@ where
if let Some(duration_since_epoch) = fetch_time() {
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.encode(),
)
.await
if let Err(e) =
kv_store.write(SCORER_NAMESPACE, "", SCORER_KEY, scorer.encode()).await
{
log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e);
// We opt not to abort early on persistence failure here as persisting
Expand Down Expand Up @@ -1080,9 +1071,9 @@ where
let fut = async {
kv_store
.write(
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
CHANNEL_MANAGER_PERSISTENCE_KEY,
CHANNEL_MANAGER_NAMESPACE,
"",
CHANNEL_MANAGER_KEY,
channel_manager.get_cm().encode(),
)
.await
Expand Down Expand Up @@ -1143,9 +1134,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_NAMESPACE,
"",
NETWORK_GRAPH_KEY,
network_graph.encode(),
)
.await
Expand Down Expand Up @@ -1183,14 +1174,8 @@ where
log_trace!(logger, "Persisting scorer");
}
let fut = async {
if let Err(e) = kv_store
.write(
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
SCORER_PERSISTENCE_KEY,
scorer.encode(),
)
.await
if let Err(e) =
kv_store.write(SCORER_NAMESPACE, "", SCORER_KEY, scorer.encode()).await
{
log_error!(
logger,
Expand Down Expand Up @@ -1301,30 +1286,18 @@ 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_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.encode(),
)
.await?;
kv_store.write(SCORER_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.encode(),
)
.write(NETWORK_GRAPH_NAMESPACE, "", NETWORK_GRAPH_KEY, network_graph.encode())
.await?;
}
Ok(())
Expand Down Expand Up @@ -1526,12 +1499,9 @@ impl BackgroundProcessor {
.expect("Time should be sometime after 1970");
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.encode(),
) {
if let Err(e) =
kv_store.write(SCORER_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 +1597,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_NAMESPACE,
"",
CHANNEL_MANAGER_KEY,
channel_manager.get_cm().encode(),
))?;
log_trace!(logger, "Done persisting ChannelManager.");
Expand Down Expand Up @@ -1666,9 +1636,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_NAMESPACE,
"",
NETWORK_GRAPH_KEY,
network_graph.encode(),
) {
log_error!(logger, "Error: Failed to persist network graph, check your disk and permissions {}", e);
Expand All @@ -1694,12 +1664,9 @@ impl BackgroundProcessor {
.expect("Time should be sometime after 1970");
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.encode(),
) {
if let Err(e) =
kv_store.write(SCORER_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 +1702,19 @@ 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_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.encode(),
)?;
kv_store.write(SCORER_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_NAMESPACE,
"",
NETWORK_GRAPH_KEY,
network_graph.encode(),
)?;
}
Expand Down Expand Up @@ -1844,12 +1806,8 @@ 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_NAMESPACE,
NETWORK_GRAPH_KEY, NETWORK_GRAPH_NAMESPACE, SCORER_KEY, SCORER_NAMESPACE,
};
use lightning::util::ser::Writeable;
use lightning::util::sweep::{
Expand Down Expand Up @@ -2104,19 +2062,15 @@ 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_NAMESPACE && key == CHANNEL_MANAGER_KEY {
assert_eq!(secondary_namespace, "");
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_NAMESPACE && key == NETWORK_GRAPH_KEY {
assert_eq!(secondary_namespace, "");
if let Some(sender) = &self.graph_persistence_notifier {
match sender.send(()) {
Ok(()) => {},
Expand All @@ -2131,10 +2085,8 @@ mod tests {
}
}

if primary_namespace == SCORER_PERSISTENCE_PRIMARY_NAMESPACE
&& secondary_namespace == SCORER_PERSISTENCE_SECONDARY_NAMESPACE
&& key == SCORER_PERSISTENCE_KEY
{
if primary_namespace == SCORER_NAMESPACE && key == SCORER_KEY {
assert_eq!(secondary_namespace, "");
if let Some((error, message)) = self.scorer_error {
return Err(std::io::Error::new(error, message).into());
}
Expand Down
13 changes: 6 additions & 7 deletions lightning-persister/src/test_utils.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ use lightning::ln::functional_test_utils::{
};
use lightning::util::persist::{
migrate_kv_store_data, read_channel_monitors, KVStoreSync, MigratableKVStore,
KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
NAMESPACE_ALPHABET, NAMESPACE_MAX_LEN,
};
use lightning::util::test_utils;
use lightning::{check_added_monitors, check_closed_broadcast, check_closed_event};
Expand Down Expand Up @@ -46,8 +46,8 @@ pub(crate) fn do_read_write_remove_list_persist<K: KVStoreSync + RefUnwindSafe>(
assert_eq!(listed_keys.len(), 0);

// Ensure we have no issue operating with primary_namespace/secondary_namespace/key being
// KVSTORE_NAMESPACE_KEY_MAX_LEN
let max_chars = "A".repeat(KVSTORE_NAMESPACE_KEY_MAX_LEN);
// NAMESPACE_MAX_LEN
let max_chars = "A".repeat(NAMESPACE_MAX_LEN);
kv_store.write(&max_chars, &max_chars, &max_chars, data.clone()).unwrap();

let listed_keys = kv_store.list(&max_chars, &max_chars).unwrap();
Expand Down Expand Up @@ -76,17 +76,16 @@ pub(crate) fn do_test_data_migration<S: MigratableKVStore, T: MigratableKVStore>
let primary_namespace = if i == 0 {
String::new()
} else {
format!("testspace{}", KVSTORE_NAMESPACE_KEY_ALPHABET.chars().nth(i).unwrap())
format!("testspace{}", NAMESPACE_ALPHABET.chars().nth(i).unwrap())
};
for j in 0..num_secondary_namespaces {
let secondary_namespace = if i == 0 || j == 0 {
String::new()
} else {
format!("testsubspace{}", KVSTORE_NAMESPACE_KEY_ALPHABET.chars().nth(j).unwrap())
format!("testsubspace{}", NAMESPACE_ALPHABET.chars().nth(j).unwrap())
};
for k in 0..num_keys {
let key =
format!("testkey{}", KVSTORE_NAMESPACE_KEY_ALPHABET.chars().nth(k).unwrap());
let key = format!("testkey{}", NAMESPACE_ALPHABET.chars().nth(k).unwrap());
source_store
.write(&primary_namespace, &secondary_namespace, &key, dummy_data.clone())
.unwrap();
Expand Down
5 changes: 2 additions & 3 deletions lightning-persister/src/utils.rs
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
use lightning::types::string::PrintableString;
use lightning::util::persist::{KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN};
use lightning::util::persist::{NAMESPACE_ALPHABET, NAMESPACE_MAX_LEN};

pub(crate) fn is_valid_kvstore_str(key: &str) -> bool {
key.len() <= KVSTORE_NAMESPACE_KEY_MAX_LEN
&& key.chars().all(|c| KVSTORE_NAMESPACE_KEY_ALPHABET.contains(c))
key.len() <= NAMESPACE_MAX_LEN && key.chars().all(|c| NAMESPACE_ALPHABET.contains(c))
}

pub(crate) fn check_namespace_key_validity(
Expand Down
Loading
Loading