Skip to content

Commit 640e4fd

Browse files
tnulltnull
committed
Add async migratable KV store migration API
Expose an async migratable KV store trait and async migration helper so async stores can migrate data without using the synchronous API. Backport of 9b11fda Co-Authored-By: HAL 9000
1 parent 2c178d1 commit 640e4fd

1 file changed

Lines changed: 143 additions & 7 deletions

File tree

‎lightning/src/util/persist.rs‎

Lines changed: 143 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -567,6 +567,128 @@ pub trait MigratableKVStoreSync: KVStoreSync {
567567
fn list_all_keys(&self) -> Result<Vec<(String, String, String)>, io::Error>;
568568
}
569569

570+
/// Provides additional interface methods that are required for [`KVStore`]-to-[`KVStore`]
571+
/// data migration.
572+
///
573+
/// This is not exported to bindings users as async is only supported in Rust.
574+
pub trait MigratableKVStore: KVStore {
575+
/// Returns *all* known keys as a list of `primary_namespace`, `secondary_namespace`, `key` tuples.
576+
///
577+
/// This is useful for migrating data from [`KVStore`] implementation to [`KVStore`]
578+
/// implementation.
579+
///
580+
/// Must exhaustively return all entries known to the store to ensure no data is missed, but
581+
/// may return the items in arbitrary order.
582+
fn list_all_keys(
583+
&self,
584+
) -> impl Future<Output = Result<Vec<(String, String, String)>, io::Error>> + 'static + MaybeSend;
585+
}
586+
587+
impl<K> MigratableKVStore for K
588+
where
589+
K: Deref,
590+
K::Target: MigratableKVStore,
591+
{
592+
fn list_all_keys(
593+
&self,
594+
) -> impl Future<Output = Result<Vec<(String, String, String)>, io::Error>> + 'static + MaybeSend
595+
{
596+
self.deref().list_all_keys()
597+
}
598+
}
599+
600+
/// This is not exported to bindings users as async is only supported in Rust.
601+
impl<K: Deref> MigratableKVStore for KVStoreSyncWrapper<K>
602+
where
603+
K::Target: MigratableKVStoreSync,
604+
{
605+
fn list_all_keys(
606+
&self,
607+
) -> impl Future<Output = Result<Vec<(String, String, String)>, io::Error>> + 'static + MaybeSend
608+
{
609+
let res = self.0.list_all_keys();
610+
611+
async move { res }
612+
}
613+
}
614+
615+
type MigrationKey = (String, String, String);
616+
617+
trait MigrationKVStore {
618+
fn list_all_keys(
619+
&self,
620+
) -> impl Future<Output = Result<Vec<MigrationKey>, io::Error>> + MaybeSend;
621+
fn read(
622+
&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
623+
) -> impl Future<Output = Result<Vec<u8>, io::Error>> + MaybeSend;
624+
fn write(
625+
&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
626+
) -> impl Future<Output = Result<(), io::Error>> + MaybeSend;
627+
}
628+
629+
struct MigrationKVStoreSyncAdapter<'a, K: ?Sized>(&'a K);
630+
631+
impl<K: MigratableKVStoreSync + ?Sized> MigrationKVStore for MigrationKVStoreSyncAdapter<'_, K> {
632+
fn list_all_keys(
633+
&self,
634+
) -> impl Future<Output = Result<Vec<MigrationKey>, io::Error>> + MaybeSend {
635+
let res = MigratableKVStoreSync::list_all_keys(self.0);
636+
637+
async move { res }
638+
}
639+
640+
fn read(
641+
&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
642+
) -> impl Future<Output = Result<Vec<u8>, io::Error>> + MaybeSend {
643+
let res = KVStoreSync::read(self.0, primary_namespace, secondary_namespace, key);
644+
645+
async move { res }
646+
}
647+
648+
fn write(
649+
&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
650+
) -> impl Future<Output = Result<(), io::Error>> + MaybeSend {
651+
let res = KVStoreSync::write(self.0, primary_namespace, secondary_namespace, key, buf);
652+
653+
async move { res }
654+
}
655+
}
656+
657+
struct MigrationKVStoreAsyncAdapter<'a, K: ?Sized>(&'a K);
658+
659+
impl<K: MigratableKVStore + ?Sized> MigrationKVStore for MigrationKVStoreAsyncAdapter<'_, K> {
660+
fn list_all_keys(
661+
&self,
662+
) -> impl Future<Output = Result<Vec<MigrationKey>, io::Error>> + MaybeSend {
663+
MigratableKVStore::list_all_keys(self.0)
664+
}
665+
666+
fn read(
667+
&self, primary_namespace: &str, secondary_namespace: &str, key: &str,
668+
) -> impl Future<Output = Result<Vec<u8>, io::Error>> + MaybeSend {
669+
KVStore::read(self.0, primary_namespace, secondary_namespace, key)
670+
}
671+
672+
fn write(
673+
&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>,
674+
) -> impl Future<Output = Result<(), io::Error>> + MaybeSend {
675+
KVStore::write(self.0, primary_namespace, secondary_namespace, key, buf)
676+
}
677+
}
678+
679+
async fn migrate_kv_store_data_inner<S: MigrationKVStore, T: MigrationKVStore>(
680+
source_store: S, target_store: T,
681+
) -> Result<(), io::Error> {
682+
let keys_to_migrate = source_store.list_all_keys().await?;
683+
684+
for (primary_namespace, secondary_namespace, key) in &keys_to_migrate {
685+
let data = source_store.read(primary_namespace, secondary_namespace, key).await?;
686+
target_store.write(primary_namespace, secondary_namespace, key, data).await?;
687+
}
688+
689+
Ok(())
690+
}
691+
570692
/// Migrates all data from one store to another.
571693
///
572694
/// This operation assumes that `target_store` is empty, i.e., any data present under copied keys
@@ -578,14 +700,28 @@ pub trait MigratableKVStoreSync: KVStoreSync {
578700
pub fn migrate_kv_store_data<S: MigratableKVStoreSync, T: MigratableKVStoreSync>(
579701
source_store: &mut S, target_store: &mut T,
580702
) -> Result<(), io::Error> {
581-
let keys_to_migrate = source_store.list_all_keys()?;
582-
583-
for (primary_namespace, secondary_namespace, key) in &keys_to_migrate {
584-
let data = source_store.read(primary_namespace, secondary_namespace, key)?;
585-
target_store.write(primary_namespace, secondary_namespace, key, data)?;
586-
}
703+
poll_sync_future(migrate_kv_store_data_inner(
704+
MigrationKVStoreSyncAdapter(source_store),
705+
MigrationKVStoreSyncAdapter(target_store),
706+
))
707+
}
587708

588-
Ok(())
709+
/// Migrates all data from one asynchronous store to another.
710+
///
711+
/// This operation assumes that `target_store` is empty, i.e., any data present under copied keys
712+
/// might get overriden. User must ensure `source_store` is not modified during operation,
713+
/// otherwise no consistency guarantees can be given.
714+
///
715+
/// Will abort and return an error if any IO operation fails. Note that in this case the
716+
/// `target_store` might get left in an intermediate state.
717+
pub async fn migrate_kv_store_data_async<S: MigratableKVStore, T: MigratableKVStore>(
718+
source_store: &S, target_store: &T,
719+
) -> Result<(), io::Error> {
720+
migrate_kv_store_data_inner(
721+
MigrationKVStoreAsyncAdapter(source_store),
722+
MigrationKVStoreAsyncAdapter(target_store),
723+
)
724+
.await
589725
}
590726

591727
impl<ChannelSigner: EcdsaChannelSigner, K: KVStoreSync + ?Sized> Persist<ChannelSigner> for K {

0 commit comments

Comments
 (0)