Optimize async I/O and locking patterns to reduce runtime blocking

devolutions-gateway/src/api/update.rs

@@ -64,11 +64,13 @@ pub(super) async fn trigger_update_check(
             .err(),
     )?;
 
-    std::fs::write(updater_file_path, update_json).map_err(
-        HttpError::internal()
-            .with_msg("failed to write the new `update.json` manifest on disk")
-            .err(),
-    )?;
+    tokio::fs::write(updater_file_path, update_json)
+        .await
+        .map_err(
+            HttpError::internal()
+                .with_msg("failed to write the new `update.json` manifest on disk")
+                .err(),
+        )?;
 
     Ok(Json(UpdateResponse {}))
 }

devolutions-gateway/src/credential.rs

@@ -5,7 +5,7 @@ use std::sync::Arc;
 use anyhow::Context;
 use async_trait::async_trait;
 use devolutions_gateway_task::{ShutdownSignal, Task};
-use parking_lot::Mutex;
+use parking_lot::RwLock;
 use serde::{de, ser};
 use uuid::Uuid;
 
@@ -27,7 +27,7 @@ pub struct AppCredentialMapping {
 }
 
 #[derive(Debug, Clone)]
-pub struct CredentialStoreHandle(Arc<Mutex<CredentialStore>>);
+pub struct CredentialStoreHandle(Arc<RwLock<CredentialStore>>);
 
 impl Default for CredentialStoreHandle {
     fn default() -> Self {
@@ -37,7 +37,7 @@ impl Default for CredentialStoreHandle {
 
 impl CredentialStoreHandle {
     pub fn new() -> Self {
-        Self(Arc::new(Mutex::new(CredentialStore::new())))
+        Self(Arc::new(RwLock::new(CredentialStore::new())))
     }
 
     pub fn insert(
@@ -46,11 +46,11 @@ impl CredentialStoreHandle {
         mapping: Option<AppCredentialMapping>,
         time_to_live: time::Duration,
     ) -> anyhow::Result<Option<ArcCredentialEntry>> {
-        self.0.lock().insert(token, mapping, time_to_live)
+        self.0.write().insert(token, mapping, time_to_live)
     }
 
     pub fn get(&self, token_id: Uuid) -> Option<ArcCredentialEntry> {
-        self.0.lock().get(token_id)
+        self.0.read().get(token_id)
     }
 }
 
@@ -205,7 +205,7 @@ async fn cleanup_task(handle: CredentialStoreHandle, mut shutdown_signal: Shutdo
 
         let now = time::OffsetDateTime::now_utc();
 
-        handle.0.lock().entries.retain(|_, src| now < src.expires_at);
+        handle.0.write().entries.retain(|_, src| now < src.expires_at);
     }
 
     debug!("Task terminated");

devolutions-gateway/src/plugin_manager/recording.rs

@@ -130,7 +130,8 @@ impl Recorder {
         path_array.truncate(path_size - 1);
 
         #[allow(clippy::cast_sign_loss)] // Losing the sign of the value on purpose.
-        let str_path = String::from_utf8(path_array.iter().map(|element| *element as u8).collect());
+        let byte_array: Vec<u8> = path_array.into_iter().map(|element| element as u8).collect();
+        let str_path = String::from_utf8(byte_array);
 
         match str_path {
             Ok(path) => Ok(PathBuf::from(path)),

devolutions-gateway/src/recording.rs

@@ -45,15 +45,15 @@ struct JrecManifest {
 }
 
 impl JrecManifest {
-    fn read_from_file(path: impl AsRef<Path>) -> anyhow::Result<Self> {
-        let json = std::fs::read(path)?;
+    async fn read_from_file(path: impl AsRef<Path>) -> anyhow::Result<Self> {
+        let json = fs::read(path).await?;
         let manifest = serde_json::from_slice(&json)?;
         Ok(manifest)
     }
 
-    fn save_to_file(&self, path: impl AsRef<Path>) -> anyhow::Result<()> {
+    async fn save_to_file(&self, path: impl AsRef<Path>) -> anyhow::Result<()> {
         let json = serde_json::to_string_pretty(&self)?;
-        std::fs::write(path, json)?;
+        fs::write(path, json).await?;
         Ok(())
     }
 }
@@ -494,7 +494,7 @@ impl RecordingManagerTask {
             debug!(path = %recording_path, "Recording directory already exists");
 
             let mut existing_manifest =
-                JrecManifest::read_from_file(&manifest_path).context("read manifest from disk")?;
+                JrecManifest::read_from_file(&manifest_path).await.context("read manifest from disk")?;
             let next_file_idx = existing_manifest.files.len();
 
             let start_time = time::OffsetDateTime::now_utc().unix_timestamp();
@@ -510,6 +510,7 @@ impl RecordingManagerTask {
 
             existing_manifest
                 .save_to_file(&manifest_path)
+                .await
                 .context("override existing manifest")?;
 
             (existing_manifest, recording_file)
@@ -539,6 +540,7 @@ impl RecordingManagerTask {
 
             initial_manifest
                 .save_to_file(&manifest_path)
+                .await
                 .context("write initial manifest to disk")?;
 
             (initial_manifest, recording_file)
@@ -616,6 +618,7 @@ impl RecordingManagerTask {
         ongoing
             .manifest
             .save_to_file(&ongoing.manifest_path)
+            .await
             .with_context(|| format!("write manifest at {}", ongoing.manifest_path))?;
 
         // Notify all the streamers that recording has ended.

devolutions-gateway/src/service.rs

@@ -252,7 +252,7 @@ async fn spawn_tasks(conf_handle: ConfHandle) -> anyhow::Result<Tasks> {
 
     let token_cache = devolutions_gateway::token::new_token_cache().pipe(Arc::new);
 
-    let jrl = load_jrl_from_disk(&conf)?;
+    let jrl = load_jrl_from_disk(&conf).await?;
 
     let (session_manager_handle, session_manager_rx) = session_manager_channel();
 
@@ -358,12 +358,13 @@ async fn spawn_tasks(conf_handle: ConfHandle) -> anyhow::Result<Tasks> {
     Ok(tasks)
 }
 
-fn load_jrl_from_disk(config: &Conf) -> anyhow::Result<Arc<CurrentJrl>> {
+async fn load_jrl_from_disk(config: &Conf) -> anyhow::Result<Arc<CurrentJrl>> {
     let jrl_file = config.jrl_file.as_path();
 
     let claims: JrlTokenClaims = if jrl_file.exists() {
         info!("Reading JRL file from disk (path: {jrl_file})");
-        std::fs::read_to_string(jrl_file)
+        tokio::fs::read_to_string(jrl_file)
+            .await
             .context("couldn't read JRL file")?
             .pipe_deref(serde_json::from_str)
             .context("invalid JRL")?