wip

src/storage/mod.rs

@@ -16,12 +16,13 @@ use std::{
     collections::{HashMap, HashSet},
     ffi::OsStr,
     fmt, fs,
+    future::Future,
     io::{self, BufReader},
     ops::RangeInclusive,
     path::{Path, PathBuf},
-    sync::Arc,
+    sync::{Arc, Mutex},
 };
-use tokio::{io::AsyncWriteExt, runtime::Runtime};
+use tokio::{io::AsyncWriteExt, runtime::Runtime, sync::RwLock};
 use tracing::{error, info_span, instrument, trace};
 
 type FileRange = RangeInclusive<u64>;
@@ -111,6 +112,7 @@ enum StorageBackend {
 pub struct AsyncStorage {
     backend: StorageBackend,
     config: Arc<Config>,
+    archive_index_locks: Mutex<HashMap<PathBuf, tokio::sync::RwLock<()>>>,
 }
 
 impl AsyncStorage {
@@ -121,6 +123,7 @@ impl AsyncStorage {
     ) -> Result<Self> {
         Ok(Self {
             config: config.clone(),
+            archive_index_locks: Mutex::new(HashMap::new()),
             backend: match config.storage_backend {
                 StorageKind::Database => {
                     StorageBackend::Database(DatabaseBackend::new(pool, metrics))
@@ -303,6 +306,22 @@ impl AsyncStorage {
         Ok(blob)
     }
 
+    // async fn with_archive_index_lock<Fut, F>(&self, index_path: impl AsRef<Path>, f: F)
+    // where
+    //     Fut: Future<Output = Result<()>> + Send,
+    //     F: Fn() -> Fut + Send + 'static,
+    // {
+    //     let mut locks = self.archive_index_locks.lock().unwrap();
+
+    //     if let Some(lock) = locks.get(index_path.as_ref()) {
+    //         f(&lock);
+    //     } else {
+    //         f(&locks
+    //             .entry(index_path.as_ref().to_path_buf())
+    //             .or_insert_with(|| RwLock::new(())));
+    //     }
+    // }
+
     #[instrument]
     pub async fn download_archive_index(
         &self,