feat: Replace LRU cache with Moka async cache in policy variables (#1166)

Co-authored-by: houseme <housemecn@gmail.com>

crates/iam/src/manager.rs

@@ -23,6 +23,7 @@ use crate::{
         UpdateServiceAccountOpts,
     },
 };
+use futures::future::join_all;
 use rustfs_ecstore::global::get_global_action_cred;
 use rustfs_madmin::{AccountStatus, AddOrUpdateUserReq, GroupDesc};
 use rustfs_policy::{
@@ -402,13 +403,25 @@ where
 
         self.cache.policy_docs.store(Arc::new(cache));
 
-        let ret = m
+        let items: Vec<_> = m.into_iter().map(|(k, v)| (k, v.policy.clone())).collect();
+
+        let futures: Vec<_> = items.iter().map(|(_, policy)| policy.match_resource(bucket_name)).collect();
+
+        let results = join_all(futures).await;
+
+        let filtered = items
             .into_iter()
-            .filter(|(_, v)| bucket_name.is_empty() || v.policy.match_resource(bucket_name))
-            .map(|(k, v)| (k, v.policy))
+            .zip(results)
+            .filter_map(|((k, policy), matches)| {
+                if bucket_name.is_empty() || matches {
+                    Some((k, policy))
+                } else {
+                    None
+                }
+            })
             .collect();
 
-        Ok(ret)
+        Ok(filtered)
     }
 
     pub async fn merge_policies(&self, name: &str) -> (String, Policy) {
@@ -456,22 +469,51 @@ where
 
         self.cache.policy_docs.store(Arc::new(cache));
 
-        let ret = m
-            .into_iter()
-            .filter(|(_, v)| bucket_name.is_empty() || v.policy.match_resource(bucket_name))
+        let items: Vec<_> = m.into_iter().map(|(k, v)| (k, v.clone())).collect();
+
+        let futures: Vec<_> = items
+            .iter()
+            .map(|(_, policy_doc)| policy_doc.policy.match_resource(bucket_name))
             .collect();
 
-        Ok(ret)
+        let results = join_all(futures).await;
+
+        let filtered = items
+            .into_iter()
+            .zip(results)
+            .filter_map(|((k, policy_doc), matches)| {
+                if bucket_name.is_empty() || matches {
+                    Some((k, policy_doc))
+                } else {
+                    None
+                }
+            })
+            .collect();
+
+        Ok(filtered)
     }
 
     pub async fn list_policy_docs_internal(&self, bucket_name: &str) -> Result<HashMap<String, PolicyDoc>> {
-        let ret = self
-            .cache
-            .policy_docs
-            .load()
+        let cache = self.cache.policy_docs.load();
+        let items: Vec<_> = cache.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
+
+        let futures: Vec<_> = items
             .iter()
-            .filter(|(_, v)| bucket_name.is_empty() || v.policy.match_resource(bucket_name))
-            .map(|(k, v)| (k.clone(), v.clone()))
+            .map(|(_, policy_doc)| policy_doc.policy.match_resource(bucket_name))
+            .collect();
+
+        let results = join_all(futures).await;
+
+        let ret = items
+            .into_iter()
+            .zip(results)
+            .filter_map(|((k, policy_doc), matches)| {
+                if bucket_name.is_empty() || matches {
+                    Some((k, policy_doc))
+                } else {
+                    None
+                }
+            })
             .collect();
 
         Ok(ret)
@@ -1753,7 +1795,7 @@ fn filter_policies(cache: &Cache, policy_name: &str, bucket_name: &str) -> (Stri
         }
 
         if let Some(p) = cache.policy_docs.load().get(&policy) {
-            if bucket_name.is_empty() || p.policy.match_resource(bucket_name) {
+            if bucket_name.is_empty() || pollster::block_on(p.policy.match_resource(bucket_name)) {
                 policies.push(policy);
                 to_merge.push(p.policy.clone());
             }

crates/iam/src/sys.rs

@@ -755,10 +755,10 @@ impl<T: Store> IamSys<T> {
 
         let (has_session_policy, is_allowed_sp) = is_allowed_by_session_policy(args);
         if has_session_policy {
-            return is_allowed_sp && (is_owner || combined_policy.is_allowed(args));
+            return is_allowed_sp && (is_owner || combined_policy.is_allowed(args).await);
         }
 
-        is_owner || combined_policy.is_allowed(args)
+        is_owner || combined_policy.is_allowed(args).await
     }
 
     pub async fn is_allowed_service_account(&self, args: &Args<'_>, parent_user: &str) -> bool {
@@ -814,15 +814,15 @@ impl<T: Store> IamSys<T> {
         };
 
         if sa_str == INHERITED_POLICY_TYPE {
-            return is_owner || combined_policy.is_allowed(&parent_args);
+            return is_owner || combined_policy.is_allowed(&parent_args).await;
         }
 
         let (has_session_policy, is_allowed_sp) = is_allowed_by_session_policy_for_service_account(args);
         if has_session_policy {
-            return is_allowed_sp && (is_owner || combined_policy.is_allowed(&parent_args));
+            return is_allowed_sp && (is_owner || combined_policy.is_allowed(&parent_args).await);
         }
 
-        is_owner || combined_policy.is_allowed(&parent_args)
+        is_owner || combined_policy.is_allowed(&parent_args).await
     }
 
     pub async fn get_combined_policy(&self, policies: &[String]) -> Policy {
@@ -857,7 +857,7 @@ impl<T: Store> IamSys<T> {
             return false;
         }
 
-        self.get_combined_policy(&policies).await.is_allowed(args)
+        self.get_combined_policy(&policies).await.is_allowed(args).await
     }
 }
 
@@ -883,7 +883,7 @@ fn is_allowed_by_session_policy(args: &Args<'_>) -> (bool, bool) {
     let mut session_policy_args = args.clone();
     session_policy_args.is_owner = false;
 
-    (has_session_policy, sub_policy.is_allowed(&session_policy_args))
+    (has_session_policy, pollster::block_on(sub_policy.is_allowed(&session_policy_args)))
 }
 
 fn is_allowed_by_session_policy_for_service_account(args: &Args<'_>) -> (bool, bool) {
@@ -909,7 +909,7 @@ fn is_allowed_by_session_policy_for_service_account(args: &Args<'_>) -> (bool, b
     let mut session_policy_args = args.clone();
     session_policy_args.is_owner = false;
 
-    (has_session_policy, sub_policy.is_allowed(&session_policy_args))
+    (has_session_policy, pollster::block_on(sub_policy.is_allowed(&session_policy_args)))
 }
 
 #[derive(Debug, Clone, Default)]