use crate::utils::{check_namespace_key_validity, is_valid_kvstore_str};
use lightning::util::persist::{KVStore, MigratableKVStore};
use lightning::util::string::PrintableString;
use std::collections::HashMap;
use std::fs;
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, RwLock};
#[cfg(target_os = "windows")]
use {std::ffi::OsStr, std::os::windows::ffi::OsStrExt};
#[cfg(target_os = "windows")]
macro_rules! call {
($e: expr) => {
if $e != 0 {
Ok(())
} else {
Err(std::io::Error::last_os_error())
}
};
}
#[cfg(target_os = "windows")]
fn path_to_windows_str<T: AsRef<OsStr>>(path: &T) -> Vec<u16> {
path.as_ref().encode_wide().chain(Some(0)).collect()
}
const GC_LOCK_INTERVAL: usize = 25;
pub struct FilesystemStore {
data_dir: PathBuf,
tmp_file_counter: AtomicUsize,
gc_counter: AtomicUsize,
locks: Mutex<HashMap<PathBuf, Arc<RwLock<()>>>>,
}
impl FilesystemStore {
pub fn new(data_dir: PathBuf) -> Self {
let locks = Mutex::new(HashMap::new());
let tmp_file_counter = AtomicUsize::new(0);
let gc_counter = AtomicUsize::new(1);
Self { data_dir, tmp_file_counter, gc_counter, locks }
}
pub fn get_data_dir(&self) -> PathBuf {
self.data_dir.clone()
}
fn garbage_collect_locks(&self) {
let gc_counter = self.gc_counter.fetch_add(1, Ordering::AcqRel);
if gc_counter % GC_LOCK_INTERVAL == 0 {
let mut outer_lock = self.locks.lock().unwrap();
outer_lock.retain(|_, v| Arc::strong_count(&v) > 1);
}
}
fn get_dest_dir_path(
&self, primary_namespace: &str, secondary_namespace: &str,
) -> std::io::Result<PathBuf> {
let mut dest_dir_path = {
#[cfg(target_os = "windows")]
{
let data_dir = self.data_dir.clone();
fs::create_dir_all(data_dir.clone())?;
fs::canonicalize(data_dir)?
}
#[cfg(not(target_os = "windows"))]
{
self.data_dir.clone()
}
};
dest_dir_path.push(primary_namespace);
if !secondary_namespace.is_empty() {
dest_dir_path.push(secondary_namespace);
}
Ok(dest_dir_path)
}
}
impl KVStore for FilesystemStore {
fn read(
&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
) -> lightning::io::Result<Vec<u8>> {
check_namespace_key_validity(primary_namespace, secondary_namespace, Some(key), "read")?;
let mut dest_file_path = self.get_dest_dir_path(primary_namespace, secondary_namespace)?;
dest_file_path.push(key);
let mut buf = Vec::new();
{
let inner_lock_ref = {
let mut outer_lock = self.locks.lock().unwrap();
Arc::clone(&outer_lock.entry(dest_file_path.clone()).or_default())
};
let _guard = inner_lock_ref.read().unwrap();
let mut f = fs::File::open(dest_file_path)?;
f.read_to_end(&mut buf)?;
}
self.garbage_collect_locks();
Ok(buf)
}
fn write(
&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8],
) -> lightning::io::Result<()> {
check_namespace_key_validity(primary_namespace, secondary_namespace, Some(key), "write")?;
let mut dest_file_path = self.get_dest_dir_path(primary_namespace, secondary_namespace)?;
dest_file_path.push(key);
let parent_directory = dest_file_path.parent().ok_or_else(|| {
let msg =
format!("Could not retrieve parent directory of {}.", dest_file_path.display());
std::io::Error::new(std::io::ErrorKind::InvalidInput, msg)
})?;
fs::create_dir_all(&parent_directory)?;
let mut tmp_file_path = dest_file_path.clone();
let tmp_file_ext = format!("{}.tmp", self.tmp_file_counter.fetch_add(1, Ordering::AcqRel));
tmp_file_path.set_extension(tmp_file_ext);
{
let mut tmp_file = fs::File::create(&tmp_file_path)?;
tmp_file.write_all(&buf)?;
tmp_file.sync_all()?;
}
let res = {
let inner_lock_ref = {
let mut outer_lock = self.locks.lock().unwrap();
Arc::clone(&outer_lock.entry(dest_file_path.clone()).or_default())
};
let _guard = inner_lock_ref.write().unwrap();
#[cfg(not(target_os = "windows"))]
{
fs::rename(&tmp_file_path, &dest_file_path)?;
let dir_file = fs::OpenOptions::new().read(true).open(&parent_directory)?;
dir_file.sync_all()?;
Ok(())
}
#[cfg(target_os = "windows")]
{
let res = if dest_file_path.exists() {
call!(unsafe {
windows_sys::Win32::Storage::FileSystem::ReplaceFileW(
path_to_windows_str(&dest_file_path).as_ptr(),
path_to_windows_str(&tmp_file_path).as_ptr(),
std::ptr::null(),
windows_sys::Win32::Storage::FileSystem::REPLACEFILE_IGNORE_MERGE_ERRORS,
std::ptr::null_mut() as *const core::ffi::c_void,
std::ptr::null_mut() as *const core::ffi::c_void,
)
})
} else {
call!(unsafe {
windows_sys::Win32::Storage::FileSystem::MoveFileExW(
path_to_windows_str(&tmp_file_path).as_ptr(),
path_to_windows_str(&dest_file_path).as_ptr(),
windows_sys::Win32::Storage::FileSystem::MOVEFILE_WRITE_THROUGH
| windows_sys::Win32::Storage::FileSystem::MOVEFILE_REPLACE_EXISTING,
)
})
};
match res {
Ok(()) => {
let dest_file =
fs::OpenOptions::new().read(true).write(true).open(&dest_file_path)?;
dest_file.sync_all()?;
Ok(())
},
Err(e) => Err(e.into()),
}
}
};
self.garbage_collect_locks();
res
}
fn remove(
&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
) -> lightning::io::Result<()> {
check_namespace_key_validity(primary_namespace, secondary_namespace, Some(key), "remove")?;
let mut dest_file_path = self.get_dest_dir_path(primary_namespace, secondary_namespace)?;
dest_file_path.push(key);
if !dest_file_path.is_file() {
return Ok(());
}
{
let inner_lock_ref = {
let mut outer_lock = self.locks.lock().unwrap();
Arc::clone(&outer_lock.entry(dest_file_path.clone()).or_default())
};
let _guard = inner_lock_ref.write().unwrap();
if lazy {
fs::remove_file(&dest_file_path)?;
} else {
#[cfg(not(target_os = "windows"))]
{
fs::remove_file(&dest_file_path)?;
let parent_directory = dest_file_path.parent().ok_or_else(|| {
let msg = format!(
"Could not retrieve parent directory of {}.",
dest_file_path.display()
);
std::io::Error::new(std::io::ErrorKind::InvalidInput, msg)
})?;
let dir_file = fs::OpenOptions::new().read(true).open(parent_directory)?;
dir_file.sync_all()?;
}
#[cfg(target_os = "windows")]
{
let mut trash_file_path = dest_file_path.clone();
let trash_file_ext =
format!("{}.trash", self.tmp_file_counter.fetch_add(1, Ordering::AcqRel));
trash_file_path.set_extension(trash_file_ext);
call!(unsafe {
windows_sys::Win32::Storage::FileSystem::MoveFileExW(
path_to_windows_str(&dest_file_path).as_ptr(),
path_to_windows_str(&trash_file_path).as_ptr(),
windows_sys::Win32::Storage::FileSystem::MOVEFILE_WRITE_THROUGH
| windows_sys::Win32::Storage::FileSystem::MOVEFILE_REPLACE_EXISTING,
)
})?;
{
let trash_file = fs::OpenOptions::new()
.read(true)
.write(true)
.open(&trash_file_path.clone())?;
trash_file.sync_all()?;
}
fs::remove_file(trash_file_path).ok();
}
}
}
self.garbage_collect_locks();
Ok(())
}
fn list(
&self, primary_namespace: &str, secondary_namespace: &str,
) -> lightning::io::Result<Vec<String>> {
check_namespace_key_validity(primary_namespace, secondary_namespace, None, "list")?;
let prefixed_dest = self.get_dest_dir_path(primary_namespace, secondary_namespace)?;
let mut keys = Vec::new();
if !Path::new(&prefixed_dest).exists() {
return Ok(Vec::new());
}
for entry in fs::read_dir(&prefixed_dest)? {
let entry = entry?;
let p = entry.path();
if !dir_entry_is_key(&p)? {
continue;
}
let key = get_key_from_dir_entry(&p, &prefixed_dest)?;
keys.push(key);
}
self.garbage_collect_locks();
Ok(keys)
}
}
fn dir_entry_is_store_artifact(path: &Path) -> bool {
match path.extension().and_then(|ext| ext.to_str()) {
Some("tmp") => true,
Some("trash") => {
#[cfg(target_os = "windows")]
{
fs::remove_file(path).ok();
}
true
},
_ => false,
}
}
fn dir_entry_is_key(p: &Path) -> Result<bool, lightning::io::Error> {
if dir_entry_is_store_artifact(&p) {
return Ok(false);
}
let metadata = p.metadata().map_err(|e| {
let msg = format!(
"Failed to list keys at path {}: {}",
PrintableString(p.to_str().unwrap_or_default()),
e
);
lightning::io::Error::new(lightning::io::ErrorKind::Other, msg)
})?;
if metadata.is_dir() {
return Ok(false);
}
if !metadata.is_file() {
debug_assert!(
false,
"Failed to list keys at path {}: file couldn't be accessed.",
PrintableString(p.to_str().unwrap_or_default())
);
let msg = format!(
"Failed to list keys at path {}: file couldn't be accessed.",
PrintableString(p.to_str().unwrap_or_default())
);
return Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, msg));
}
Ok(true)
}
fn get_key_from_dir_entry(p: &Path, base_path: &Path) -> Result<String, lightning::io::Error> {
match p.strip_prefix(&base_path) {
Ok(stripped_path) => {
if let Some(relative_path) = stripped_path.to_str() {
if is_valid_kvstore_str(relative_path) {
return Ok(relative_path.to_string());
} else {
debug_assert!(
false,
"Failed to list keys of path {}: file path is not valid key",
PrintableString(p.to_str().unwrap_or_default())
);
let msg = format!(
"Failed to list keys of path {}: file path is not valid key",
PrintableString(p.to_str().unwrap_or_default())
);
return Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, msg));
}
} else {
debug_assert!(
false,
"Failed to list keys of path {}: file path is not valid UTF-8",
PrintableString(p.to_str().unwrap_or_default())
);
let msg = format!(
"Failed to list keys of path {}: file path is not valid UTF-8",
PrintableString(p.to_str().unwrap_or_default())
);
return Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, msg));
}
},
Err(e) => {
debug_assert!(
false,
"Failed to list keys of path {}: {}",
PrintableString(p.to_str().unwrap_or_default()),
e
);
let msg = format!(
"Failed to list keys of path {}: {}",
PrintableString(p.to_str().unwrap_or_default()),
e
);
return Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, msg));
},
}
}
impl MigratableKVStore for FilesystemStore {
fn list_all_keys(&self) -> Result<Vec<(String, String, String)>, lightning::io::Error> {
let prefixed_dest = &self.data_dir;
if !prefixed_dest.exists() {
return Ok(Vec::new());
}
let mut keys = Vec::new();
'primary_loop: for primary_entry in fs::read_dir(prefixed_dest)? {
let primary_path = primary_entry?.path();
if dir_entry_is_store_artifact(&primary_path) {
continue 'primary_loop;
}
if dir_entry_is_key(&primary_path)? {
let primary_namespace = String::new();
let secondary_namespace = String::new();
let key = get_key_from_dir_entry(&primary_path, prefixed_dest)?;
keys.push((primary_namespace, secondary_namespace, key));
continue 'primary_loop;
}
'secondary_loop: for secondary_entry in fs::read_dir(&primary_path)? {
let secondary_path = secondary_entry?.path();
if dir_entry_is_store_artifact(&secondary_path) {
continue 'secondary_loop;
}
if dir_entry_is_key(&secondary_path)? {
let primary_namespace = get_key_from_dir_entry(&primary_path, prefixed_dest)?;
let secondary_namespace = String::new();
let key = get_key_from_dir_entry(&secondary_path, &primary_path)?;
keys.push((primary_namespace, secondary_namespace, key));
continue 'secondary_loop;
}
for tertiary_entry in fs::read_dir(&secondary_path)? {
let tertiary_entry = tertiary_entry?;
let tertiary_path = tertiary_entry.path();
if dir_entry_is_store_artifact(&tertiary_path) {
continue;
}
if dir_entry_is_key(&tertiary_path)? {
let primary_namespace =
get_key_from_dir_entry(&primary_path, prefixed_dest)?;
let secondary_namespace =
get_key_from_dir_entry(&secondary_path, &primary_path)?;
let key = get_key_from_dir_entry(&tertiary_path, &secondary_path)?;
keys.push((primary_namespace, secondary_namespace, key));
} else {
debug_assert!(
false,
"Failed to list keys of path {}: only two levels of namespaces are supported",
PrintableString(tertiary_path.to_str().unwrap_or_default())
);
let msg = format!(
"Failed to list keys of path {}: only two levels of namespaces are supported",
PrintableString(tertiary_path.to_str().unwrap_or_default())
);
return Err(lightning::io::Error::new(
lightning::io::ErrorKind::Other,
msg,
));
}
}
}
}
Ok(keys)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::test_utils::{
do_read_write_remove_list_persist, do_test_data_migration, do_test_store,
};
use bitcoin::Txid;
use lightning::chain::chainmonitor::Persist;
use lightning::chain::transaction::OutPoint;
use lightning::chain::ChannelMonitorUpdateStatus;
use lightning::check_closed_event;
use lightning::events::{ClosureReason, MessageSendEventsProvider};
use lightning::ln::functional_test_utils::*;
use lightning::util::persist::read_channel_monitors;
use lightning::util::test_utils;
use std::str::FromStr;
impl Drop for FilesystemStore {
fn drop(&mut self) {
match fs::remove_dir_all(&self.data_dir) {
Err(e) => println!("Failed to remove test persister directory: {}", e),
_ => {},
}
}
}
#[test]
fn read_write_remove_list_persist() {
let mut temp_path = std::env::temp_dir();
temp_path.push("test_read_write_remove_list_persist");
let fs_store = FilesystemStore::new(temp_path);
do_read_write_remove_list_persist(&fs_store);
}
#[test]
fn list_all_keys_skips_leftover_store_artifacts() {
let mut temp_path = std::env::temp_dir();
temp_path.push("test_list_all_keys_skips_leftover_store_artifacts");
let fs_store = FilesystemStore::new(temp_path.clone());
KVStore::write(&fs_store, "primary", "secondary", "key", &[1]).unwrap();
fs::write(temp_path.join("top_level.0.tmp"), b"stale").unwrap();
fs::write(temp_path.join("top_level.0.trash"), b"stale").unwrap();
let primary_path = temp_path.join("primary");
fs::write(primary_path.join("primary_level.0.tmp"), b"stale").unwrap();
fs::write(primary_path.join("primary_level.0.trash"), b"stale").unwrap();
let secondary_path = primary_path.join("secondary");
fs::write(secondary_path.join("secondary_level.0.tmp"), b"stale").unwrap();
fs::write(secondary_path.join("secondary_level.0.trash"), b"stale").unwrap();
let keys = fs_store.list_all_keys().unwrap();
assert_eq!(keys, vec![("primary".to_string(), "secondary".to_string(), "key".to_string())]);
}
#[test]
fn test_data_migration() {
let mut source_temp_path = std::env::temp_dir();
source_temp_path.push("test_data_migration_source");
let mut source_store = FilesystemStore::new(source_temp_path);
let mut target_temp_path = std::env::temp_dir();
target_temp_path.push("test_data_migration_target");
let mut target_store = FilesystemStore::new(target_temp_path);
do_test_data_migration(&mut source_store, &mut target_store);
}
#[test]
fn test_if_monitors_is_not_dir() {
let store = FilesystemStore::new("test_monitors_is_not_dir".into());
fs::create_dir_all(&store.get_data_dir()).unwrap();
let mut path = std::path::PathBuf::from(&store.get_data_dir());
path.push("monitors");
fs::File::create(path).unwrap();
let chanmon_cfgs = create_chanmon_cfgs(1);
let mut node_cfgs = create_node_cfgs(1, &chanmon_cfgs);
let chain_mon_0 = test_utils::TestChainMonitor::new(
Some(&chanmon_cfgs[0].chain_source),
&chanmon_cfgs[0].tx_broadcaster,
&chanmon_cfgs[0].logger,
&chanmon_cfgs[0].fee_estimator,
&store,
node_cfgs[0].keys_manager,
);
node_cfgs[0].chain_monitor = chain_mon_0;
let node_chanmgrs = create_node_chanmgrs(1, &node_cfgs, &[None]);
let nodes = create_network(1, &node_cfgs, &node_chanmgrs);
assert!(
read_channel_monitors(&store, nodes[0].keys_manager, nodes[0].keys_manager).is_err()
);
}
#[test]
fn test_filesystem_store() {
let store_0 = FilesystemStore::new("test_filesystem_store_0".into());
let store_1 = FilesystemStore::new("test_filesystem_store_1".into());
do_test_store(&store_0, &store_1)
}
#[cfg(not(target_os = "windows"))]
#[test]
fn test_readonly_dir_perm_failure() {
let store = FilesystemStore::new("test_readonly_dir_perm_failure".into());
fs::create_dir_all(&store.get_data_dir()).unwrap();
let chanmon_cfgs = create_chanmon_cfgs(2);
let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
let chan = create_announced_chan_between_nodes(&nodes, 0, 1);
let error_message = "Channel force-closed";
nodes[1]
.node
.force_close_broadcasting_latest_txn(
&chan.2,
&nodes[0].node.get_our_node_id(),
error_message.to_string(),
)
.unwrap();
check_closed_event!(
nodes[1],
1,
ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(true) },
[nodes[0].node.get_our_node_id()],
100000
);
let mut added_monitors = nodes[1].chain_monitor.added_monitors.lock().unwrap();
let path = &store.get_data_dir();
let mut perms = fs::metadata(path).unwrap().permissions();
perms.set_readonly(true);
fs::set_permissions(path, perms).unwrap();
let test_txo = OutPoint {
txid: Txid::from_str(
"8984484a580b825b9972d7adb15050b3ab624ccd731946b3eeddb92f4e7ef6be",
)
.unwrap(),
index: 0,
};
match store.persist_new_channel(test_txo, &added_monitors[0].1) {
ChannelMonitorUpdateStatus::UnrecoverableError => {},
_ => panic!("unexpected result from persisting new channel"),
}
nodes[1].node.get_and_clear_pending_msg_events();
added_monitors.clear();
}
#[cfg(target_os = "windows")]
#[test]
fn test_fail_on_open() {
let chanmon_cfgs = create_chanmon_cfgs(2);
let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
let chan = create_announced_chan_between_nodes(&nodes, 0, 1);
let error_message = "Channel force-closed";
nodes[1]
.node
.force_close_broadcasting_latest_txn(
&chan.2,
&nodes[0].node.get_our_node_id(),
error_message.to_string(),
)
.unwrap();
check_closed_event!(
nodes[1],
1,
ClosureReason::HolderForceClosed { broadcasted_latest_txn: Some(true) },
[nodes[0].node.get_our_node_id()],
100000
);
let mut added_monitors = nodes[1].chain_monitor.added_monitors.lock().unwrap();
let update_map = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap();
let update_id = update_map.get(&added_monitors[0].1.channel_id()).unwrap();
let store = FilesystemStore::new(":<>/".into());
let test_txo = OutPoint {
txid: Txid::from_str(
"8984484a580b825b9972d7adb15050b3ab624ccd731946b3eeddb92f4e7ef6be",
)
.unwrap(),
index: 0,
};
match store.persist_new_channel(test_txo, &added_monitors[0].1) {
ChannelMonitorUpdateStatus::UnrecoverableError => {},
_ => panic!("unexpected result from persisting new channel"),
}
nodes[1].node.get_and_clear_pending_msg_events();
added_monitors.clear();
}
}
#[cfg(ldk_bench)]
pub mod bench {
use criterion::Criterion;
pub fn bench_sends(bench: &mut Criterion) {
let store_a = super::FilesystemStore::new("bench_filesystem_store_a".into());
let store_b = super::FilesystemStore::new("bench_filesystem_store_b".into());
lightning::ln::channelmanager::bench::bench_two_sends(
bench,
"bench_filesystem_persisted_sends",
store_a,
store_b,
);
}
}