WIP 4

crates/cdk/src/wallet/key_manager/mod.rs

@@ -132,8 +132,11 @@ impl Drop for SharedWorker {
     }
 }
 
+/// Type alias for the worker registry to simplify the complex type
+type WorkerRegistry = Arc<ParkingLotRwLock<HashMap<MintUrl, Arc<SharedWorker>>>>;
+
 /// Global registry of workers, one per mint_url
-static WORKER_REGISTRY: Lazy<Arc<ParkingLotRwLock<HashMap<MintUrl, Arc<SharedWorker>>>>> =
+static WORKER_REGISTRY: Lazy<WorkerRegistry> =
     Lazy::new(|| Arc::new(ParkingLotRwLock::new(HashMap::new())));
 
 /// Key manager for a single mint

crates/cdk/src/wallet/key_manager/worker.rs

@@ -21,13 +21,17 @@ use crate::Error;
 #[cfg(feature = "auth")]
 use crate::wallet::AuthMintConnector;
 
+/// Type alias for storage list to improve readability
+type StorageList =
+    Arc<ParkingLotRwLock<Vec<Arc<dyn WalletDatabase<Err = database::Error> + Send + Sync>>>>;
+
 /// Messages for the background refresh task
 #[derive(Debug, Clone)]
 pub(super) enum MessageToWorker {
     /// Stop the refresh task
     Stop,
 
-    ///
+    /// Sync a new storage backend with the current cache
     SyncDb(Arc<dyn WalletDatabase<Err = database::Error> + Send + Sync>),
 
     /// Fetch keys from the mint immediately
@@ -39,13 +43,9 @@ impl KeyManager {
     ///
     /// First checks all databases for the keyset. If not found,
     /// fetches from the mint server via HTTP and persists to all databases.
-    pub(super) async fn load_keyset_from_db_or_http(
+    async fn load_keyset_from_db_or_http(
         mint_url: &MintUrl,
-        storages: &Arc<
-            ParkingLotRwLock<
-                Vec<Arc<dyn WalletDatabase<Err = cdk_common::database::Error> + Send + Sync>>,
-            >,
-        >,
+        storages: &StorageList,
         client: &Arc<dyn MintConnector + Send + Sync>,
         keyset_id: &Id,
     ) -> Result<KeySet, Error> {
@@ -95,98 +95,148 @@ impl KeyManager {
         Ok(keyset)
     }
 
-    /// Load mint info and keys from all databases
+    /// Load cached mint data from all registered storage backends
     ///
-    /// Iterates through all storages and loads keysets and keys into cache.
-    /// This is called on first access when cache is empty.
-    pub(super) async fn fetch_mint_info_and_keys_from_db(
+    /// Iterates through all storages, collecting unique keysets and keys.
+    /// Marks cache as ready only if mint_info was found.
+    ///
+    /// Returns a MintKeyCache that may or may not be ready depending on what was found.
+    async fn load_cache_from_storages(
         mint_url: &MintUrl,
-        storages: &Arc<
-            ParkingLotRwLock<
-                Vec<Arc<dyn WalletDatabase<Err = cdk_common::database::Error> + Send + Sync>>,
-            >,
-        >,
+        storages: &StorageList,
     ) -> Result<MintKeyCache, Error> {
-        tracing::debug!("Cache empty, loading from storage first for {}", mint_url);
+        tracing::debug!("Loading cache from storage for {}", mint_url);
 
-        let mut storage_cache = MintKeyCache::empty();
+        let mut cache = MintKeyCache::empty();
         let storages_list = storages.read().clone();
 
         for storage in storages_list {
-            if storage_cache.mint_info.is_none() {
-                storage_cache.mint_info = storage.get_mint(mint_url.clone()).await?;
+            // Load mint info from first storage that has it
+            if cache.mint_info.is_none() {
+                cache.mint_info = storage.get_mint(mint_url.clone()).await?;
             }
 
+            // Collect unique keysets from all storages
             for keyset in storage
                 .get_mint_keysets(mint_url.clone())
                 .await?
                 .unwrap_or_default()
             {
-                if storage_cache.keysets_by_id.contains_key(&keyset.id) {
+                if cache.keysets_by_id.contains_key(&keyset.id) {
                     continue;
                 }
 
                 let arc_keyset = Arc::new(keyset.clone());
-                storage_cache
-                    .keysets_by_id
-                    .insert(keyset.id, arc_keyset.clone());
+                cache.keysets_by_id.insert(keyset.id, arc_keyset.clone());
 
                 if keyset.active {
-                    storage_cache.active_keysets.push(arc_keyset);
+                    cache.active_keysets.push(arc_keyset);
                 }
             }
 
-            for keyset_id in storage_cache.keysets_by_id.keys() {
-                if storage_cache.keys_by_id.contains_key(keyset_id) {
+            // Collect unique keys from all storages
+            for keyset_id in cache.keysets_by_id.keys() {
+                if cache.keys_by_id.contains_key(keyset_id) {
                     continue;
                 }
 
                 if let Some(keys) = storage.get_keys(keyset_id).await? {
-                    storage_cache.keys_by_id.insert(*keyset_id, Arc::new(keys));
+                    cache.keys_by_id.insert(*keyset_id, Arc::new(keys));
                 }
             }
         }
 
-        if storage_cache.mint_info.is_some() {
-            storage_cache.refresh_version += 1;
-            storage_cache.is_ready = true;
+        // Mark as ready only if we found mint info
+        if cache.mint_info.is_some() {
+            cache.refresh_version += 1;
+            cache.is_ready = true;
         }
 
         tracing::debug!(
             "Loaded {} keys from storage for {}",
-            storage_cache.keys_by_id.len(),
+            cache.keys_by_id.len(),
             mint_url
         );
 
-        Ok(storage_cache)
+        Ok(cache)
+    }
+
+    /// Finalize cache state with proper versioning and timestamp
+    ///
+    /// Sets the cache as ready and updates metadata.
+    fn finalize_cache(cache: &mut MintKeyCache, previous_version: u64) {
+        cache.refresh_version = previous_version + 1;
+        cache.is_ready = true;
+        cache.last_refresh = std::time::Instant::now();
+    }
+
+    /// Write keys that don't already exist in storage
+    ///
+    /// Checks each key in cache and only writes if not already in storage.
+    async fn write_missing_keys_to_storage(
+        storage: &Arc<dyn WalletDatabase<Err = database::Error> + Send + Sync>,
+        mint_url: &MintUrl,
+        cache: &MintKeyCache,
+    ) {
+        for (keyset_id, keys) in cache.keys_by_id.iter() {
+            // Skip if already exists
+            if storage.get_keys(keyset_id).await.ok().flatten().is_some() {
+                continue;
+            }
+
+            // Get keyset info
+            let Some(keyset_info) = cache.keysets_by_id.get(keyset_id) else {
+                tracing::warn!("Missing keyset info for {}", keyset_id);
+                continue;
+            };
+
+            // Write keyset
+            let keyset = KeySet {
+                id: *keyset_id,
+                unit: keyset_info.unit.clone(),
+                final_expiry: keyset_info.final_expiry,
+                keys: (**keys).clone(),
+            };
+
+            if let Err(e) = storage.add_keys(keyset).await {
+                tracing::warn!(
+                    "Failed to write keys for {} to {}: {}",
+                    keyset_id,
+                    mint_url,
+                    e
+                );
+            }
+        }
     }
 
-    /// Persist cache to a single database
+    /// Write cached data to a single storage backend
     ///
-    /// Writes mint info, keysets, and keys to the given storage backend.
-    /// Errors are logged but don't fail the operation.
-    pub(super) async fn persist_cache_db(
-        storage: Arc<dyn WalletDatabase<Err = cdk_common::database::Error> + Send + Sync>,
+    /// Persists mint_info, keysets, and keys. Only writes keys that don't already exist.
+    /// All errors are logged but don't stop the operation.
+    async fn write_cache_to_storage(
+        storage: Arc<dyn WalletDatabase<Err = database::Error> + Send + Sync>,
         mint_url: MintUrl,
-        new_cache: Arc<MintKeyCache>,
+        cache: Arc<MintKeyCache>,
     ) {
-        if !new_cache.is_ready {
+        if !cache.is_ready {
             return;
         }
 
-        if new_cache.mint_info.is_some() {
+        // Write mint info
+        if cache.mint_info.is_some() {
             let _ = storage
-                .add_mint(mint_url.clone(), new_cache.mint_info.clone())
+                .add_mint(mint_url.clone(), cache.mint_info.clone())
                 .await
                 .inspect_err(|e| {
                     tracing::warn!("Failed to persist mint_info for {}: {}", mint_url, e);
                 });
         }
 
+        // Write keysets
         let _ = storage
             .add_mint_keysets(
                 mint_url.clone(),
-                new_cache
+                cache
                     .keysets_by_id
                     .values()
                     .map(|ks| (**ks).clone())
@@ -195,83 +245,88 @@ impl KeyManager {
             .await
             .inspect_err(|e| tracing::warn!("Failed to persist keysets for {}: {}", mint_url, e));
 
-        for (keyset_id, keys) in new_cache.keys_by_id.iter() {
-            if storage
-                .get_keys(keyset_id)
-                .await
-                .inspect_err(|e| tracing::warn!("Failed to get_keys {e}"))
-                .unwrap_or_default()
-                .is_none()
-            {
-                let keyset = if let Some(v) = new_cache.keysets_by_id.get(keyset_id) {
-                    v
-                } else {
-                    tracing::warn!("Malformed keysets, cannot find {}", keyset_id);
-                    continue;
-                };
-                let _ = storage
-                    .add_keys(KeySet {
-                        id: *keyset_id,
-                        unit: keyset.unit.clone(),
-                        final_expiry: keyset.final_expiry,
-                        keys: (**keys).clone(),
-                    })
-                    .await
-                    .inspect_err(|e| {
-                        tracing::warn!("Failed to persist keys for keyset {}: {}", keyset_id, e)
-                    });
-            }
-        }
+        // Write missing keys
+        Self::write_missing_keys_to_storage(&storage, &mint_url, &cache).await;
     }
 
-    /// Persist cache to all registered databases
+    /// Write cached data to all registered storage backends concurrently
     ///
-    /// Spawns a task for each storage backend to write cache asynchronously.
-    pub(super) async fn persist_cache(
-        storages: &Arc<
-            ParkingLotRwLock<
-                Vec<Arc<dyn WalletDatabase<Err = cdk_common::database::Error> + Send + Sync>>,
-            >,
-        >,
+    /// Spawns a task for each storage to write in parallel.
+    async fn write_cache_to_all_storages(
+        storages: &StorageList,
         mint_url: MintUrl,
-        new_cache: Arc<MintKeyCache>,
+        cache: Arc<MintKeyCache>,
     ) {
         let storages_list = storages.read().clone();
 
         for storage in storages_list {
-            spawn(Self::persist_cache_db(
+            spawn(Self::write_cache_to_storage(
                 storage,
                 mint_url.clone(),
-                new_cache.clone(),
+                cache.clone(),
             ));
         }
     }
 
-    /// Fetch keys from mint server via HTTP
+    /// Attempt to load cache from storages if not ready, and persist if successful
     ///
-    /// Fetches mint info, keysets, and keys from the mint server. Updates cache
-    /// and schedules next refresh. Persists new data to all databases.
-    pub(super) async fn fetch_from_http(
+    /// Returns true if cache is ready after this operation.
+    async fn try_load_cache_from_storages(
+        mint_url: &MintUrl,
+        storages: &StorageList,
+        cache: &Arc<ArcSwap<MintKeyCache>>,
+    ) -> bool {
+        // Already ready, nothing to do
+        if cache.load().is_ready {
+            return true;
+        }
+
+        // Try loading from storages
+        let loaded_cache = match Self::load_cache_from_storages(mint_url, storages).await {
+            Ok(c) => c,
+            Err(e) => {
+                tracing::warn!("Failed to load cache from storage for {}: {}", mint_url, e);
+                return false;
+            }
+        };
+
+        // If we got valid data, persist and store
+        if loaded_cache.is_ready {
+            let loaded_cache = Arc::new(loaded_cache);
+            Self::write_cache_to_all_storages(storages, mint_url.clone(), loaded_cache.clone())
+                .await;
+            cache.store(loaded_cache);
+            return true;
+        }
+
+        false
+    }
+
+    /// Fetch fresh mint data from HTTP endpoint
+    ///
+    /// 1. Fetches mint info and validates time sync
+    /// 2. Fetches keyset list (including auth keysets if enabled)
+    /// 3. Loads actual keyset keys (trying storage first, HTTP fallback)
+    /// 4. Updates shared cache and schedules next refresh
+    /// 5. Persists all data to all registered storages
+    pub(super) async fn fetch_mint_data_from_http(
         mint_url: MintUrl,
         client: Arc<dyn MintConnector + Send + Sync>,
         #[cfg(feature = "auth")] auth_client: Arc<dyn AuthMintConnector + Send + Sync>,
-        storages: Arc<
-            ParkingLotRwLock<
-                Vec<Arc<dyn WalletDatabase<Err = cdk_common::database::Error> + Send + Sync>>,
-            >,
-        >,
+        storages: StorageList,
         cache: Arc<ArcSwap<MintKeyCache>>,
         refresh_scheduler: RefreshScheduler,
     ) -> Result<(), Error> {
         tracing::debug!("Fetching keys from mint server for {}", mint_url);
 
+        // Fetch and validate mint info
         let mint_info = client.get_mint_info().await?;
 
         if let Some(mint_unix_time) = mint_info.time {
             let current_unix_time = unix_time();
             if current_unix_time.abs_diff(mint_unix_time) > 30 {
                 tracing::warn!(
-                    "Mint time does match wallet time. Mint: {}, Wallet: {}",
+                    "Mint time does not match wallet time. Mint: {}, Wallet: {}",
                     mint_unix_time,
                     current_unix_time
                 );
@@ -279,20 +334,17 @@ impl KeyManager {
             }
         }
 
+        // Fetch keysets
         let keysets_response = client.get_mint_keysets().await?;
+        let mut keysets = keysets_response.keysets;
 
-        let keysets = keysets_response.keysets;
-
+        // Include auth keysets if enabled
         #[cfg(feature = "auth")]
-        let keysets =
-            if let Ok(auth_keysets_response) = auth_client.get_mint_blind_auth_keysets().await {
-                let mut keysets = keysets;
-                keysets.extend_from_slice(&auth_keysets_response.keysets);
-                keysets
-            } else {
-                keysets
-            };
+        if let Ok(auth_keysets_response) = auth_client.get_mint_blind_auth_keysets().await {
+            keysets.extend_from_slice(&auth_keysets_response.keysets);
+        }
 
+        // Build new cache
         let mut new_cache = MintKeyCache::empty();
 
         for keyset_info in keysets {
@@ -305,7 +357,7 @@ impl KeyManager {
                 new_cache.active_keysets.push(arc_keyset);
             }
 
-            // Try to load keyset from database first, then HTTP
+            // Load keyset keys (try database first, then HTTP)
             if let Ok(keyset) =
                 Self::load_keyset_from_db_or_http(&mint_url, &storages, &client, &keyset_info.id)
                     .await
@@ -323,13 +375,13 @@ impl KeyManager {
             }
         }
 
+        // Schedule next refresh
         refresh_scheduler.schedule_refresh(mint_url.clone());
 
+        // Finalize cache
         let old_generation = cache.load().refresh_version;
         new_cache.mint_info = Some(mint_info);
-        new_cache.refresh_version = old_generation + 1;
-        new_cache.is_ready = true;
-        new_cache.last_refresh = std::time::Instant::now();
+        Self::finalize_cache(&mut new_cache, old_generation);
 
         tracing::debug!(
             "Refreshed {} keysets and {} keys for {} (generation {})",
@@ -339,45 +391,36 @@ impl KeyManager {
             new_cache.refresh_version
         );
 
+        // Persist and update cache
         let new_cache = Arc::new(new_cache);
-        Self::persist_cache(&storages, mint_url, new_cache.clone()).await;
+        Self::write_cache_to_all_storages(&storages, mint_url, new_cache.clone()).await;
         cache.store(new_cache);
 
-        Ok::<(), Error>(())
+        Ok(())
     }
 
-    /// Refresh keys from mint server
+    /// Spawn task to refresh mint data (database-first, then HTTP)
     ///
-    /// Spawns an async task with 60s timeout.
-    pub(super) fn fetch_and_sync_mint_task(
+    /// First attempts to load from storage. If that fails or cache is still not ready,
+    /// fetches from HTTP with a 60s timeout. All operations run in spawned task.
+    pub(super) fn refresh_mint_task(
         mint_url: MintUrl,
         client: Arc<dyn MintConnector + Send + Sync>,
         #[cfg(feature = "auth")] auth_client: Arc<dyn AuthMintConnector + Send + Sync>,
-        storages: Arc<
-            ParkingLotRwLock<
-                Vec<Arc<dyn WalletDatabase<Err = cdk_common::database::Error> + Send + Sync>>,
-            >,
-        >,
+        storages: StorageList,
         cache: Arc<ArcSwap<MintKeyCache>>,
         refresh_scheduler: RefreshScheduler,
     ) {
         spawn(async move {
-            if !cache.load().is_ready {
-                if let Ok(new_cache) = Self::fetch_mint_info_and_keys_from_db(&mint_url, &storages)
-                    .await
-                    .inspect_err(|e| {
-                        tracing::warn!("Failed to load keys from storage for {}: {}", mint_url, e)
-                    })
-                {
-                    let new_cache = Arc::new(new_cache);
-                    Self::persist_cache(&storages, mint_url.clone(), new_cache.clone()).await;
-                    cache.store(new_cache);
-                }
+            // Try loading from storage first
+            if Self::try_load_cache_from_storages(&mint_url, &storages, &cache).await {
+                return;
             }
 
+            // If still not ready, fetch from HTTP
             let result = tokio::time::timeout(
                 Duration::from_secs(60),
-                Self::fetch_from_http(
+                Self::fetch_mint_data_from_http(
                     mint_url.clone(),
                     client,
                     #[cfg(feature = "auth")]
@@ -389,25 +432,19 @@ impl KeyManager {
             )
             .await;
 
+            // Log result
             let _ = result
                 .map_err(|_| Error::Timeout)
                 .and_then(|r| r)
                 .inspect(|_| {
-                    tracing::debug!(
-                        "Successfully fetched keys from mint server for {}",
-                        mint_url
-                    );
+                    tracing::debug!("Successfully refreshed keys for {}", mint_url);
                 })
                 .inspect_err(|e| match e {
                     Error::Timeout => {
                         tracing::error!("Timeout fetching keys from mint server for {}", mint_url)
                     }
                     _ => {
-                        tracing::error!(
-                            "Failed to fetch keys from mint server for {}: {}",
-                            mint_url,
-                            e
-                        )
+                        tracing::error!("Failed to refresh keys for {}: {}", mint_url, e)
                     }
                 });
         });
@@ -422,11 +459,7 @@ impl KeyManager {
         mint_url: MintUrl,
         client: Arc<dyn MintConnector + Send + Sync>,
         #[cfg(feature = "auth")] auth_client: Arc<dyn AuthMintConnector + Send + Sync>,
-        storages: Arc<
-            ParkingLotRwLock<
-                Vec<Arc<dyn WalletDatabase<Err = cdk_common::database::Error> + Send + Sync>>,
-            >,
-        >,
+        storages: StorageList,
         cache: Arc<ArcSwap<MintKeyCache>>,
         refresh_interval: Duration,
     ) {
@@ -443,13 +476,18 @@ impl KeyManager {
                             break;
                         }
                         MessageToWorker::SyncDb(db) => {
-                            let cache_for_spawn = cache.clone();
-                            let mint_for_spawn = mint_url.clone();
-                            tokio::spawn(async move {
+                            // Wait for cache to be ready, then sync this storage
+                            let cache_clone = cache.clone();
+                            let mint_clone = mint_url.clone();
+                            spawn(async move {
                                 loop {
-                                    let cache = cache_for_spawn.load();
-                                    if cache.is_ready {
-                                        Self::persist_cache_db(db, mint_for_spawn, cache.clone()).await;
+                                    let current_cache = cache_clone.load();
+                                    if current_cache.is_ready {
+                                        Self::write_cache_to_storage(
+                                            db,
+                                            mint_clone,
+                                            current_cache.clone()
+                                        ).await;
                                         break;
                                     }
                                     sleep(Duration::from_millis(100)).await;
@@ -458,7 +496,7 @@ impl KeyManager {
                         }
                         MessageToWorker::FetchMint => {
                             tracing::debug!("FetchMint message received for {}", mint_url);
-                            Self::fetch_and_sync_mint_task(
+                            Self::refresh_mint_task(
                                 mint_url.clone(),
                                 client.clone(),
                                 #[cfg(feature = "auth")]
@@ -476,7 +514,7 @@ impl KeyManager {
 
                     if !due_mints.is_empty() || !is_scheduled {
                         tracing::debug!("Time to refresh mint: {}", mint_url);
-                        Self::fetch_and_sync_mint_task(
+                        Self::refresh_mint_task(
                             mint_url.clone(),
                             client.clone(),
                             #[cfg(feature = "auth")]