Skip to content

Commit 9db060e

Browse files
tnulltnull
committed
Remove blocking KV store support
Drop the remaining synchronous KV store trait bounds and implementations. After the preceding migrations, custom stores only need to provide async KVStore persistence. Co-Authored-By: HAL 9000
1 parent 9d04aa8 commit 9db060e

11 files changed

Lines changed: 383 additions & 748 deletions

File tree

‎src/builder.rs‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,7 @@ use crate::tx_broadcaster::TransactionBroadcaster;
8181
use crate::types::{
8282
AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreRef, DynStoreWrapper,
8383
GossipSync, Graph, HRNResolver, KeysManager, MessageRouter, OnionMessenger, PaymentStore,
84-
PeerManager, PendingPaymentStore, SyncAndAsyncKVStore,
84+
PeerManager, PendingPaymentStore,
8585
};
8686
use crate::wallet::persist::KVStoreWalletPersister;
8787
use crate::wallet::Wallet;
@@ -176,17 +176,17 @@ pub enum BuildError {
176176
RuntimeSetupFailed,
177177
/// We failed to read data from the [`KVStore`].
178178
///
179-
/// [`KVStore`]: lightning::util::persist::KVStoreSync
179+
/// [`KVStore`]: lightning::util::persist::KVStore
180180
ReadFailed,
181181
/// We failed to write data to the [`KVStore`].
182182
///
183-
/// [`KVStore`]: lightning::util::persist::KVStoreSync
183+
/// [`KVStore`]: lightning::util::persist::KVStore
184184
WriteFailed,
185185
/// We failed to access the given `storage_dir_path`.
186186
StoragePathAccessFailed,
187187
/// We failed to setup our [`KVStore`].
188188
///
189-
/// [`KVStore`]: lightning::util::persist::KVStoreSync
189+
/// [`KVStore`]: lightning::util::persist::KVStore
190190
KVStoreSetupFailed,
191191
/// We failed to setup the onchain wallet.
192192
WalletSetupFailed,
@@ -826,7 +826,7 @@ impl NodeBuilder {
826826
}
827827

828828
/// Builds a [`Node`] instance according to the options previously configured.
829-
pub fn build_with_store<S: SyncAndAsyncKVStore + Send + Sync + 'static>(
829+
pub fn build_with_store<S: KVStore + Send + Sync + 'static>(
830830
&self, node_entropy: NodeEntropy, kv_store: S,
831831
) -> Result<Node, BuildError> {
832832
let logger = setup_logger(&self.log_writer_config, &self.config)?;
@@ -845,14 +845,14 @@ impl NodeBuilder {
845845
}
846846
}
847847

848-
fn build_with_store_and_logger<S: SyncAndAsyncKVStore + Send + Sync + 'static>(
848+
fn build_with_store_and_logger<S: KVStore + Send + Sync + 'static>(
849849
&self, node_entropy: NodeEntropy, kv_store: S, logger: Arc<Logger>,
850850
) -> Result<Node, BuildError> {
851851
let runtime = self.setup_runtime(&logger)?;
852852
self.build_with_store_runtime_and_logger(node_entropy, kv_store, runtime, logger)
853853
}
854854

855-
fn build_with_store_runtime_and_logger<S: SyncAndAsyncKVStore + Send + Sync + 'static>(
855+
fn build_with_store_runtime_and_logger<S: KVStore + Send + Sync + 'static>(
856856
&self, node_entropy: NodeEntropy, kv_store: S, runtime: Arc<Runtime>, logger: Arc<Logger>,
857857
) -> Result<Node, BuildError> {
858858
let seed_bytes = node_entropy.to_seed_bytes();
@@ -1346,7 +1346,7 @@ impl ArcedNodeBuilder {
13461346
/// Builds a [`Node`] instance according to the options previously configured.
13471347
// Note that the generics here don't actually work for Uniffi, but we don't currently expose
13481348
// this so its not needed.
1349-
pub fn build_with_store<S: SyncAndAsyncKVStore + Send + Sync + 'static>(
1349+
pub fn build_with_store<S: KVStore + Send + Sync + 'static>(
13501350
&self, node_entropy: Arc<NodeEntropy>, kv_store: S,
13511351
) -> Result<Arc<Node>, BuildError> {
13521352
self.inner.read().expect("lock").build_with_store(*node_entropy, kv_store).map(Arc::new)

‎src/chain/electrum.rs‎

Lines changed: 44 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -129,37 +129,6 @@ impl ElectrumChainSource {
129129
let incremental_sync =
130130
self.node_metrics.read().expect("lock").latest_onchain_wallet_sync_timestamp.is_some();
131131

132-
macro_rules! apply_wallet_update {
133-
($update_res:expr, $now:expr) => {
134-
match $update_res {
135-
Ok(update) => match onchain_wallet.apply_update(update) {
136-
Ok(()) => {
137-
log_debug!(
138-
self.logger,
139-
"{} of on-chain wallet finished in {}ms.",
140-
if incremental_sync { "Incremental sync" } else { "Sync" },
141-
$now.elapsed().as_millis()
142-
);
143-
let unix_time_secs_opt = SystemTime::now()
144-
.duration_since(UNIX_EPOCH)
145-
.ok()
146-
.map(|d| d.as_secs());
147-
update_and_persist_node_metrics(
148-
&self.node_metrics,
149-
&*self.kv_store,
150-
&*self.logger,
151-
|m| m.latest_onchain_wallet_sync_timestamp = unix_time_secs_opt,
152-
)
153-
.await?;
154-
Ok(())
155-
},
156-
Err(e) => Err(e),
157-
},
158-
Err(e) => Err(e),
159-
}
160-
};
161-
}
162-
163132
let cached_txs = onchain_wallet.get_cached_txs();
164133

165134
let res = if incremental_sync {
@@ -169,19 +138,61 @@ impl ElectrumChainSource {
169138

170139
let now = Instant::now();
171140
let update_res: Result<BdkUpdate, Error> = incremental_sync_fut.await.map(|u| u.into());
172-
apply_wallet_update!(update_res, now)
141+
self.apply_onchain_wallet_update(
142+
onchain_wallet.as_ref(),
143+
incremental_sync,
144+
update_res,
145+
now,
146+
)
147+
.await
173148
} else {
174149
let full_scan_request = onchain_wallet.get_full_scan_request();
175150
let full_scan_fut =
176151
electrum_client.get_full_scan_wallet_update(full_scan_request, cached_txs);
177152
let now = Instant::now();
178153
let update_res: Result<BdkUpdate, Error> = full_scan_fut.await.map(|u| u.into());
179-
apply_wallet_update!(update_res, now)
154+
self.apply_onchain_wallet_update(
155+
onchain_wallet.as_ref(),
156+
incremental_sync,
157+
update_res,
158+
now,
159+
)
160+
.await
180161
};
181162

182163
res
183164
}
184165

166+
async fn apply_onchain_wallet_update(
167+
&self, onchain_wallet: &Wallet, incremental_sync: bool,
168+
update_res: Result<BdkUpdate, Error>, now: Instant,
169+
) -> Result<(), Error> {
170+
match update_res {
171+
Ok(update) => match onchain_wallet.apply_update(update) {
172+
Ok(()) => {
173+
log_debug!(
174+
self.logger,
175+
"{} of on-chain wallet finished in {}ms.",
176+
if incremental_sync { "Incremental sync" } else { "Sync" },
177+
now.elapsed().as_millis()
178+
);
179+
let unix_time_secs_opt =
180+
SystemTime::now().duration_since(UNIX_EPOCH).ok().map(|d| d.as_secs());
181+
update_and_persist_node_metrics(
182+
&self.node_metrics,
183+
&*self.kv_store,
184+
&*self.logger,
185+
|m| m.latest_onchain_wallet_sync_timestamp = unix_time_secs_opt,
186+
)
187+
.await?;
188+
Ok(())
189+
},
190+
Err(e) => Err(e),
191+
},
192+
Err(e) => Err(e),
193+
}
194+
}
195+
185196
pub(crate) async fn sync_lightning_wallet(
186197
&self, channel_manager: Arc<ChannelManager>, chain_monitor: Arc<ChainMonitor>,
187198
output_sweeper: Arc<Sweeper>,

‎src/io/in_memory_store.rs‎

Lines changed: 1 addition & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -11,9 +11,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
1111
use std::sync::Mutex;
1212

1313
use lightning::io;
14-
use lightning::util::persist::{
15-
KVStore, KVStoreSync, PageToken, PaginatedKVStore, PaginatedKVStoreSync, PaginatedListResponse,
16-
};
14+
use lightning::util::persist::{KVStore, PageToken, PaginatedKVStore, PaginatedListResponse};
1715

1816
const IN_MEMORY_PAGE_SIZE: usize = 50;
1917

@@ -128,30 +126,6 @@ impl KVStore for InMemoryStore {
128126
}
129127
}
130128

131-
impl KVStoreSync for InMemoryStore {
132-
fn read(
133-
&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
134-
) -> io::Result<Vec<u8>> {
135-
self.read_internal(primary_namespace, secondary_namespace, key)
136-
}
137-
138-
fn write(
139-
&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
140-
) -> io::Result<()> {
141-
self.write_internal(primary_namespace, secondary_namespace, key, buf)
142-
}
143-
144-
fn remove(
145-
&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool,
146-
) -> io::Result<()> {
147-
self.remove_internal(primary_namespace, secondary_namespace, key, lazy)
148-
}
149-
150-
fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> io::Result<Vec<String>> {
151-
self.list_internal(primary_namespace, secondary_namespace)
152-
}
153-
}
154-
155129
impl InMemoryStore {
156130
fn list_paginated_internal(
157131
&self, primary_namespace: &str, secondary_namespace: &str, page_token: Option<PageToken>,
@@ -202,14 +176,6 @@ impl InMemoryStore {
202176
}
203177
}
204178

205-
impl PaginatedKVStoreSync for InMemoryStore {
206-
fn list_paginated(
207-
&self, primary_namespace: &str, secondary_namespace: &str, page_token: Option<PageToken>,
208-
) -> io::Result<PaginatedListResponse> {
209-
self.list_paginated_internal(primary_namespace, secondary_namespace, page_token)
210-
}
211-
}
212-
213179
impl PaginatedKVStore for InMemoryStore {
214180
fn list_paginated(
215181
&self, primary_namespace: &str, secondary_namespace: &str, page_token: Option<PageToken>,

0 commit comments

Comments
 (0)