@@ -3,9 +3,10 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use std::{
+    collections::HashMap,
     ffi::OsStr,
     path::{Component, Path, PathBuf},
-    sync::Arc,
+    sync::{Arc, Mutex, PoisonError, Weak},
     time::Duration,
 };
 
@@ -26,7 +27,7 @@ use ra_tls::attestation::{
 };
 use serde::{Deserialize, Serialize};
 use sha2::{Digest as _, Sha256};
-use tokio::io::AsyncWriteExt;
+use tokio::{io::AsyncWriteExt, sync::Mutex as AsyncMutex};
 use tpm_qvl::verify::VerifiedReport as TpmVerifiedReport;
 use tracing::{debug, info, warn};
 
@@ -121,6 +122,7 @@ pub struct CvmVerifier {
     pub download_url: String,
     pub download_timeout: Duration,
     pub attestation_verifier: Arc<AttestationVerifier>,
+    download_locks: Mutex<HashMap<String, Weak<AsyncMutex<()>>>>,
 }
 
 impl CvmVerifier {
@@ -135,9 +137,26 @@ impl CvmVerifier {
             download_url,
             download_timeout,
             attestation_verifier,
+            download_locks: Mutex::default(),
         }
     }
 
+    /// One lock per image, held from the cache check until the image is
+    /// installed, so concurrent requests for an image download it once.
+    fn download_lock(&self, hex_os_image_hash: &str) -> Arc<AsyncMutex<()>> {
+        let mut locks = self
+            .download_locks
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner);
+        locks.retain(|_, lock| lock.strong_count() > 0);
+        if let Some(lock) = locks.get(hex_os_image_hash).and_then(Weak::upgrade) {
+            return lock;
+        }
+        let lock = Arc::default();
+        locks.insert(hex_os_image_hash.to_owned(), Arc::downgrade(&lock));
+        lock
+    }
+
     fn measurement_cache_dir(&self) -> PathBuf {
         Path::new(&self.image_cache_dir).join("measurements")
     }
@@ -1022,6 +1041,10 @@ impl CvmVerifier {
     }
 
     pub async fn download_image(&self, hex_os_image_hash: &str, dst_dir: &Path) -> Result<()> {
+        let guard = self.download_lock(hex_os_image_hash).lock_owned().await;
+        if dst_dir.join("metadata.json").exists() {
+            return Ok(());
+        }
         let url = self
             .download_url
             .replace("{OS_IMAGE_HASH}", hex_os_image_hash);
@@ -1067,26 +1090,38 @@ impl CvmVerifier {
                 .context("Failed to write chunk to file")?;
         }
 
-        let extracted_dir = tmp_dir.join("extracted");
-        fs_err::create_dir_all(&extracted_dir).context("Failed to create extraction directory")?;
-
         file.flush()
             .await
             .context("Failed to flush image archive")?;
         drop(file);
-        Self::extract_image_archive(&tarball_path, &extracted_dir)?;
 
+        let hex_os_image_hash = hex_os_image_hash.to_owned();
+        let dst_dir = dst_dir.to_owned();
+        tokio::task::spawn_blocking(move || {
+            // A blocking task outlives its caller: if the caller is cancelled
+            // (e.g. by `download_timeout`), this still runs to the end. Keep
+            // the lock until the install finishes, or the next download for
+            // this image would race it on `dst_dir`.
+            let _guard = guard;
+            Self::install_image(auto_delete_temp_dir.path(), &hex_os_image_hash, &dst_dir)
+        })
+        .await
+        .context("image install task failed")?
+    }
+
+    fn install_image(tmp_dir: &Path, hex_os_image_hash: &str, dst_dir: &Path) -> Result<()> {
+        let extracted_dir = tmp_dir.join("extracted");
+        fs_err::create_dir_all(&extracted_dir).context("Failed to create extraction directory")?;
+        Self::extract_image_archive(&tmp_dir.join("image.tar.gz"), &extracted_dir)?;
         Self::verify_extracted_image(&extracted_dir, hex_os_image_hash)?;
 
         if dst_dir.exists() {
             fs_err::remove_dir_all(dst_dir).context("Failed to remove destination directory")?;
         }
         let dst_dir_parent = dst_dir.parent().context("Failed to get parent directory")?;
         fs_err::create_dir_all(dst_dir_parent).context("Failed to create parent directory")?;
-        // Move the extracted files to the destination directory
         fs_err::rename(extracted_dir, dst_dir)
-            .context("Failed to move extracted files to destination directory")?;
-        Ok(())
+            .context("Failed to move extracted files to destination directory")
     }
 }
 
@@ -1853,6 +1888,80 @@ mod tests {
         }
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn concurrent_requests_download_an_image_once() {
+        let metadata = br#"{"cmdline":"","kernel":"bzImage","initrd":"initrd","bios":"ovmf.fd"}"#;
+        let files: &[(&str, &[u8])] = &[
+            ("metadata.json", metadata),
+            ("bzImage", b"kernel"),
+            ("initrd", b"initrd"),
+            ("ovmf.fd", b"firmware"),
+        ];
+        let files_doc: String = files
+            .iter()
+            .map(|(name, payload)| format!("{}  {name}\n", hex::encode(Sha256::digest(payload))))
+            .collect();
+        let hash = hex::encode(Sha256::digest(&files_doc));
+        let mut archive = tar::Builder::new(flate2::write::GzEncoder::new(
+            Vec::new(),
+            flate2::Compression::default(),
+        ));
+        for (name, payload) in files
+            .iter()
+            .chain([&("sha256sum.txt", files_doc.as_bytes())])
+        {
+            let mut header = tar::Header::new_gnu();
+            header.set_size(payload.len() as u64);
+            header.set_mode(0o644);
+            archive.append_data(&mut header, name, *payload).unwrap();
+        }
+        let tarball = archive.into_inner().unwrap().finish().unwrap();
+
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let url = format!(
+            "http://{}/{{OS_IMAGE_HASH}}",
+            listener.local_addr().unwrap()
+        );
+        let served = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        tokio::spawn({
+            let served = served.clone();
+            async move {
+                use tokio::io::AsyncReadExt;
+                while let Ok((mut stream, _)) = listener.accept().await {
+                    served.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                    let _ = stream.read(&mut [0; 4096]).await;
+                    tokio::time::sleep(Duration::from_millis(100)).await;
+                    let head = format!(
+                        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
+                        tarball.len()
+                    );
+                    let _ = stream.write_all(head.as_bytes()).await;
+                    let _ = stream.write_all(&tarball).await;
+                }
+            }
+        });
+
+        let cache = tempfile::tempdir().unwrap();
+        let verifier = Arc::new(CvmVerifier::new(
+            cache.path().to_str().unwrap().into(),
+            url,
+            Duration::from_secs(10),
+            test_attestation_verifier(),
+        ));
+        let dst_dir = verifier.images_dir().join(&hash);
+        let downloads: Vec<_> = (0..8)
+            .map(|_| {
+                let (verifier, hash, dst_dir) = (verifier.clone(), hash.clone(), dst_dir.clone());
+                tokio::spawn(async move { verifier.download_image(&hash, &dst_dir).await })
+            })
+            .collect();
+        for download in downloads {
+            download.await.unwrap().unwrap();
+        }
+        assert_eq!(served.load(std::sync::atomic::Ordering::SeqCst), 1);
+        assert_eq!(fs_err::read(dst_dir.join("bzImage")).unwrap(), b"kernel");
+    }
+
     #[test]
     fn image_archive_rejects_links_and_accepts_regular_files() {
         let directory = tempfile::tempdir().unwrap();