Skip to content

Commit 464e6b7

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 e3ae2b2 commit 464e6b7

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
@@ -52,7 +52,7 @@ use lightning::routing::scoring::{
5252
use lightning::sign::{EntropySource, NodeSigner};
5353

5454
use lightning::util::persist::{
55-
read_channel_monitors, CHANNEL_MANAGER_PERSISTENCE_KEY,
55+
read_channel_monitors, KVStoreSync, CHANNEL_MANAGER_PERSISTENCE_KEY,
5656
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
5757
};
5858
use lightning::util::ser::ReadableArgs;
@@ -1429,7 +1429,8 @@ fn build_with_store_internal(
14291429

14301430
// Initialize the ChannelManager
14311431
let channel_manager = {
1432-
if let Ok(res) = kv_store.read(
1432+
if let Ok(res) = KVStoreSync::read(
1433+
&*kv_store,
14331434
CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE,
14341435
CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE,
14351436
CHANNEL_MANAGER_PERSISTENCE_KEY,
@@ -1667,7 +1668,7 @@ fn build_with_store_internal(
16671668
Ok(output_sweeper) => Arc::new(output_sweeper),
16681669
Err(e) => {
16691670
if e.kind() == std::io::ErrorKind::NotFound {
1670-
Arc::new(OutputSweeper::new_with_kv_store_sync(
1671+
Arc::new(OutputSweeper::new(
16711672
channel_manager.current_best_block(),
16721673
Arc::clone(&tx_broadcaster),
16731674
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
@@ -43,6 +43,7 @@ use lightning::util::config::{
4343
ChannelConfigOverrides, ChannelConfigUpdate, ChannelHandshakeConfigUpdate,
4444
};
4545
use lightning::util::errors::APIError;
46+
use lightning::util::persist::KVStoreSync;
4647
use lightning::util::ser::{Readable, ReadableArgs, Writeable, Writer};
4748

4849
use lightning_types::payment::{PaymentHash, PaymentPreimage};
@@ -356,24 +357,24 @@ where
356357

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

16281629
// Check we can read back what we persisted.
1629-
let persisted_bytes = store
1630-
.read(
1631-
EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE,
1632-
EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
1633-
EVENT_QUEUE_PERSISTENCE_KEY,
1634-
)
1635-
.unwrap();
1630+
let persisted_bytes = KVStoreSync::read(
1631+
&*store,
1632+
EVENT_QUEUE_PERSISTENCE_PRIMARY_NAMESPACE,
1633+
EVENT_QUEUE_PERSISTENCE_SECONDARY_NAMESPACE,
1634+
EVENT_QUEUE_PERSISTENCE_KEY,
1635+
)
1636+
.unwrap();
16361637
let deser_event_queue =
16371638
EventQueue::read(&mut &persisted_bytes[..], (Arc::clone(&store), logger)).unwrap();
16381639
assert_eq!(deser_event_queue.wait_next_event(), expected_event);

0 commit comments

Comments
 (0)