lexe-ln: read initial migrations concurrently in main fetch block

Author: phlip9

lexe-ln/src/migrations.rs

@@ -3,10 +3,14 @@
 //! Similar to database migrations but for VFS-persisted state. Each migration
 //! creates an empty marker file in `migrations/<name>` when complete.
 
-use std::collections::HashSet;
+use std::{collections::HashSet, future::Future, pin::Pin, sync::Arc};
 
-use anyhow::Context;
+use anyhow::{Context, format_err};
 use bytes::Bytes;
+use futures::{
+    FutureExt,
+    future::{self, BoxFuture},
+};
 use lexe_api::{
     types::Empty,
     vfs::{self, Vfs, VfsDirectory, VfsFileId},
@@ -31,8 +35,20 @@ pub struct Migrations {
 }
 
 impl Migrations {
+    /// Return a future that reads migrations once and shares the result with
+    /// all awaiters.
+    pub fn read_once<'vfs>(
+        vfs: &'vfs (impl Vfs + Send + Sync),
+    ) -> MigrationsReadOnce<'vfs> {
+        let fut = Migrations::read(vfs)
+            .map(|res| res.map(Arc::new).map_err(Arc::new))
+            .boxed()
+            .shared();
+        MigrationsReadOnce { fut }
+    }
+
     /// Read all applied migrations from the VFS.
-    pub async fn read(vfs: &impl Vfs) -> anyhow::Result<Self> {
+    async fn read(vfs: &(impl Vfs + Send + Sync)) -> anyhow::Result<Self> {
         let dir = VfsDirectory::new(vfs::MIGRATIONS_DIR);
         let dir_list = vfs
             .list_directory(&dir)
@@ -67,3 +83,24 @@ impl Migrations {
         self.applied.contains(name)
     }
 }
+
+/// A future that reads migrations once and shares the result with all awaiters.
+#[derive(Clone)]
+pub struct MigrationsReadOnce<'vfs> {
+    fut: future::Shared<
+        BoxFuture<'vfs, Result<Arc<Migrations>, Arc<anyhow::Error>>>,
+    >,
+}
+
+impl<'vfs> Future for MigrationsReadOnce<'vfs> {
+    type Output = anyhow::Result<Arc<Migrations>>;
+
+    fn poll(
+        self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Self::Output> {
+        Pin::new(&mut self.get_mut().fut)
+            .poll(cx)
+            .map_err(|err| format_err!("Could not read migrations: {err:#}"))
+    }
+}

lexe-ln/src/persister.rs

@@ -28,7 +28,7 @@ use tracing::{info, warn};
 use crate::{
     alias::LexeChainMonitorType,
     event::EventId,
-    migrations::{self, Migrations},
+    migrations::{self, Migrations, MigrationsReadOnce},
     payments::{
         PaymentMetadata, PaymentV2, PaymentWithMetadata,
         manager::{CheckedPayment, PersistedPayment},
@@ -583,12 +583,13 @@ pub trait LexePersisterMethods: Vfs {
     ///
     /// Idempotent: Creates a marker file at `migrations/payments_v2` once
     /// complete, and skips the migration if the marker file already exists.
-    #[tracing::instrument(skip_all, name = "(migrate-payments-v2)")]
+    // #[tracing::instrument(skip_all, name = "(migrate-payments-v2)")]
     async fn migrate_to_payments_v2(
         &self,
-        initial_migrations: &Migrations,
+        initial_migrations_fut: MigrationsReadOnce<'_>,
     ) -> anyhow::Result<()> {
         // Check if migration has already run
+        let initial_migrations = initial_migrations_fut.await?;
         if initial_migrations.is_applied(migrations::MARKER_PAYMENTS_V2) {
             return Ok(());
         }

node/src/run.rs

@@ -329,9 +329,8 @@ impl UserNode {
             shutdown.clone(),
         ));
 
-        // Read initial applied migrations
-        // TODO(phlip9): fetch migrations concurrently in main fetch block
-        let initial_migrations = Migrations::read(&*persister).await?;
+        // Future that reads initial applied migrations
+        let initial_migrations_fut = Migrations::read_once(&*persister);
 
         // A future which reads the approved versions list
         let read_maybe_approved_versions = persister::read_approved_versions(
@@ -341,10 +340,11 @@ impl UserNode {
         );
 
         // Fetch pending payments to initialize payments manager with
+        let initial_migrations_fut_clone = initial_migrations_fut.clone();
         let pending_payments_fut = async {
             // But first, migrate to payments v2 if needed
             persister
-                .migrate_to_payments_v2(&initial_migrations)
+                .migrate_to_payments_v2(initial_migrations_fut_clone)
                 .await
                 .context("payments_v2 migration failed")?;
 
@@ -357,20 +357,23 @@ impl UserNode {
         // Read as much as possible concurrently to reduce init time
         #[rustfmt::skip] // Does not respect 80 char line width
         let (
+            try_initial_migrations,
             try_maybe_approved_versions,
             try_maybe_changeset,
             try_existing_scids,
             try_pending_payments,
             try_maybe_revocable_clients,
             try_channel_monitor_bytes,
         ) = tokio::join!(
+            initial_migrations_fut,
             read_maybe_approved_versions,
             persister.read_wallet_changeset(),
             persister.read_scids(),
             pending_payments_fut,
             persister.read_json::<RevocableClients>(&REVOCABLE_CLIENTS_FILE_ID),
             persister.fetch_channel_monitor_bytes(),
         );
+        let initial_migrations = try_initial_migrations?;
         if deploy_env.is_staging_or_prod() {
             // Erroring here prevents an attacker with access to a target user's
             // gdrive from deleting the user's approved versions list in an