Skip to content

Commit 8c2ff8f

Browse files
tnulltnull
committed
Require both types of KVStore
As an intermediary step, we require any store to implement both `KVStore` and `KVStoreSync`, allowing us to switch over step-by-step. We already switch to the fully-async background processor variant here.
1 parent 3c7cf02 commit 8c2ff8f

10 files changed

Lines changed: 236 additions & 183 deletions

File tree

‎Cargo.toml‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@ default = []
3232
#lightning-types = { version = "0.2.0" }
3333
#lightning-invoice = { version = "0.33.0", features = ["std"] }
3434
#lightning-net-tokio = { version = "0.1.0" }
35-
#lightning-persister = { version = "0.1.0" }
35+
#lightning-persister = { version = "0.1.0", features = ["tokio"] }
3636
#lightning-background-processor = { version = "0.1.0" }
3737
#lightning-rapid-gossip-sync = { version = "0.1.0" }
3838
#lightning-block-sync = { version = "0.1.0", features = ["rest-client", "rpc-client", "tokio"] }
@@ -44,7 +44,7 @@ default = []
4444
#lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main" }
4545
#lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main", features = ["std"] }
4646
#lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main" }
47-
#lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main" }
47+
#lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main", features = ["tokio"] }
4848
#lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main" }
4949
#lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main" }
5050
#lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", branch = "main", features = ["rest-client", "rpc-client", "tokio"] }
@@ -56,7 +56,7 @@ lightning = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "
5656
lightning-types = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3e21ba37a133977d4247e86f25df983b39326994" }
5757
lightning-invoice = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3e21ba37a133977d4247e86f25df983b39326994", features = ["std"] }
5858
lightning-net-tokio = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3e21ba37a133977d4247e86f25df983b39326994" }
59-
lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3e21ba37a133977d4247e86f25df983b39326994" }
59+
lightning-persister = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3e21ba37a133977d4247e86f25df983b39326994", features = ["tokio"] }
6060
lightning-background-processor = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3e21ba37a133977d4247e86f25df983b39326994" }
6161
lightning-rapid-gossip-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3e21ba37a133977d4247e86f25df983b39326994" }
6262
lightning-block-sync = { git = "https://github.com/lightningdevkit/rust-lightning", rev = "3e21ba37a133977d4247e86f25df983b39326994", features = ["rest-client", "rpc-client", "tokio"] }
@@ -68,7 +68,7 @@ lightning-macros = { git = "https://github.com/lightningdevkit/rust-lightning",
6868
#lightning-types = { path = "../rust-lightning/lightning-types" }
6969
#lightning-invoice = { path = "../rust-lightning/lightning-invoice", features = ["std"] }
7070
#lightning-net-tokio = { path = "../rust-lightning/lightning-net-tokio" }
71-
#lightning-persister = { path = "../rust-lightning/lightning-persister" }
71+
#lightning-persister = { path = "../rust-lightning/lightning-persister", features = ["tokio"] }
7272
#lightning-background-processor = { path = "../rust-lightning/lightning-background-processor" }
7373
#lightning-rapid-gossip-sync = { path = "../rust-lightning/lightning-rapid-gossip-sync" }
7474
#lightning-block-sync = { path = "../rust-lightning/lightning-block-sync", features = ["rest-client", "rpc-client", "tokio"] }

‎src/builder.rs‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@ use lightning::routing::scoring::{
5151
use lightning::sign::{EntropySource, NodeSigner};
5252

5353
use lightning::util::persist::{
54-
read_channel_monitors, CHANNEL_MANAGER_PERSISTENCE_KEY,
54+
read_channel_monitors, KVStoreSync, CHANNEL_MANAGER_PERSISTENCE_KEY,
5555
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
5656
};
5757
use lightning::util::ser::ReadableArgs;
@@ -1387,7 +1387,8 @@ fn build_with_store_internal(
13871387

13881388
// Initialize the ChannelManager
13891389
let channel_manager = {
1390-
if let Ok(res) = kv_store.read(
1390+
if let Ok(res) = KVStoreSync::read(
1391+
&*kv_store,
13911392
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
13921393
CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
13931394
CHANNEL_MANAGER_PERSISTENCE_KEY,
@@ -1610,7 +1611,7 @@ fn build_with_store_internal(
16101611
Ok(output_sweeper) => Arc::new(output_sweeper),
16111612
Err(e) => {
16121613
if e.kind() == std::io::ErrorKind::NotFound {
1613-
Arc::new(OutputSweeper::new_with_kv_store_sync(
1614+
Arc::new(OutputSweeper::new(
16141615
channel_manager.current_best_block(),
16151616
Arc::clone(&tx_broadcaster),
16161617
Arc::clone(&fee_estimator),

‎src/data_store.rs‎

Lines changed: 31 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ use crate::logger::{log_error, LdkLogger};
99
use crate::types::DynStore;
1010
use crate::Error;
1111

12+
use lightning::util::persist::KVStoreSync;
1213
use lightning::util::ser::{Readable, Writeable};
1314

1415
use std::collections::hash_map;
@@ -98,19 +99,24 @@ where
9899
let removed = self.objects.lock().unwrap().remove(id).is_some();
99100
if removed {
100101
let store_key = id.encode_to_hex_str();
101-
self.kv_store
102-
.remove(&self.primary_namespace, &self.secondary_namespace, &store_key, false)
103-
.map_err(|e| {
104-
log_error!(
105-
self.logger,
106-
"Removing object data for key {}/{}/{} failed due to: {}",
107-
&self.primary_namespace,
108-
&self.secondary_namespace,
109-
store_key,
110-
e
111-
);
112-
Error::PersistenceFailed
113-
})?;
102+
KVStoreSync::remove(
103+
&*self.kv_store,
104+
&self.primary_namespace,
105+
&self.secondary_namespace,
106+
&store_key,
107+
false,
108+
)
109+
.map_err(|e| {
110+
log_error!(
111+
self.logger,
112+
"Removing object data for key {}/{}/{} failed due to: {}",
113+
&self.primary_namespace,
114+
&self.secondary_namespace,
115+
store_key,
116+
e
117+
);
118+
Error::PersistenceFailed
119+
})?;
114120
}
115121
Ok(())
116122
}
@@ -142,9 +148,14 @@ where
142148
fn persist(&self, object: &SO) -> Result<(), Error> {
143149
let store_key = object.id().encode_to_hex_str();
144150
let data = object.encode();
145-
self.kv_store
146-
.write(&self.primary_namespace, &self.secondary_namespace, &store_key, data)
147-
.map_err(|e| {
151+
KVStoreSync::write(
152+
&*self.kv_store,
153+
&self.primary_namespace,
154+
&self.secondary_namespace,
155+
&store_key,
156+
data,
157+
)
158+
.map_err(|e| {
148159
log_error!(
149160
self.logger,
150161
"Write for key {}/{}/{} failed due to: {}",
@@ -243,13 +254,15 @@ mod tests {
243254
let store_key = id.encode_to_hex_str();
244255

245256
// Check we start empty.
246-
assert!(store.read(&primary_namespace, &secondary_namespace, &store_key).is_err());
257+
assert!(KVStoreSync::read(&*store, &primary_namespace, &secondary_namespace, &store_key)
258+
.is_err());
247259

248260
// Check we successfully store an object and return `false`
249261
let object = TestObject { id, data: [23u8; 3] };
250262
assert_eq!(Ok(false), data_store.insert(object.clone()));
251263
assert_eq!(Some(object), data_store.get(&id));
252-
assert!(store.read(&primary_namespace, &secondary_namespace, &store_key).is_ok());
264+
assert!(KVStoreSync::read(&*store, &primary_namespace, &secondary_namespace, &store_key)
265+
.is_ok());
253266

254267
// Test re-insertion returns `true`
255268
let mut override_object = object.clone();

‎src/event.rs‎

Lines changed: 23 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ use lightning::util::config::{
4242
ChannelConfigOverrides, ChannelConfigUpdate, ChannelHandshakeConfigUpdate,
4343
};
4444
use lightning::util::errors::APIError;
45+
use lightning::util::persist::KVStoreSync;
4546
use lightning::util::ser::{Readable, ReadableArgs, Writeable, Writer};
4647

4748
use lightning_types::payment::{PaymentHash, PaymentPreimage};
@@ -355,24 +356,24 @@ where
355356

356357
fn persist_queue(&self, locked_queue: &VecDeque<Event>) -> Result<(), Error> {
357358
let data = EventQueueSerWrapper(locked_queue).encode();
358-
self.kv_store
359-
.write(
359+
KVStoreSync::write(
360+
&*self.kv_store,
361+
EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE,
362+
EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
363+
EVENT_QUEUE_PERSISTENCE_KEY,
364+
data,
365+
)
366+
.map_err(|e| {
367+
log_error!(
368+
self.logger,
369+
"Write for key {}/{}/{} failed due to: {}",
360370
EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE,
361371
EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
362372
EVENT_QUEUE_PERSISTENCE_KEY,
363-
data,
364-
)
365-
.map_err(|e| {
366-
log_error!(
367-
self.logger,
368-
"Write for key {}/{}/{} failed due to: {}",
369-
EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE,
370-
EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
371-
EVENT_QUEUE_PERSISTENCE_KEY,
372-
e
373-
);
374-
Error::PersistenceFailed
375-
})?;
373+
e
374+
);
375+
Error::PersistenceFailed
376+
})?;
376377
Ok(())
377378
}
378379
}
@@ -1598,13 +1599,13 @@ mod tests {
15981599
}
15991600

16001601
// Check we can read back what we persisted.
1601-
let persisted_bytes = store
1602-
.read(
1603-
EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE,
1604-
EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
1605-
EVENT_QUEUE_PERSISTENCE_KEY,
1606-
)
1607-
.unwrap();
1602+
let persisted_bytes = KVStoreSync::read(
1603+
&*store,
1604+
EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE,
1605+
EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
1606+
EVENT_QUEUE_PERSISTENCE_KEY,
1607+
)
1608+
.unwrap();
16081609
let deser_event_queue =
16091610
EventQueue::read(&mut &persisted_bytes[..], (Arc::clone(&store), logger)).unwrap();
16101611
assert_eq!(deser_event_queue.wait_next_event(), expected_event);

0 commit comments

Comments
 (0)