Actively manage local archive-index cache

crates/bin/docs_rs_watcher/src/db/delete.rs

@@ -36,7 +36,8 @@ pub async fn delete_crate(
         // remove existing local archive index files.
         let local_index_folder = storage
             .config()
-            .local_archive_cache_path
+            .archive_index_cache
+            .path
             .join(&remote_folder);
         if local_index_folder.exists() {
             fs::remove_dir_all(&local_index_folder)
@@ -76,7 +77,7 @@ pub async fn delete_version(
             .await?;
     }
 
-    let local_archive_cache = &storage.config().local_archive_cache_path;
+    let local_archive_cache = &storage.config().archive_index_cache.path;
     let mut paths = vec![source_archive_path(name, version)];
     if is_library {
         paths.push(rustdoc_archive_path(name, version));
@@ -498,7 +499,8 @@ mod tests {
             assert!(
                 !storage
                     .config()
-                    .local_archive_cache_path
+                    .archive_index_cache
+                    .path
                     .join(&archive_index)
                     .exists()
             );

crates/lib/docs_rs_storage/Cargo.toml

@@ -7,9 +7,8 @@ edition.workspace = true
 [features]
 testing = [
     "dep:rand",
-    "dep:docs_rs_logging",
+    "dep:dashmap",
     "docs_rs_config/testing",
-    "docs_rs_logging/testing",
     "docs_rs_opentelemetry/testing",
 ]
 
@@ -22,11 +21,10 @@ aws-sdk-s3 = "1.3.0"
 aws-smithy-types-convert = { version = "0.60.0", features = ["convert-chrono"] }
 bzip2 = "0.6.0"
 chrono = { workspace = true }
-dashmap = "6.0.0"
+dashmap = { version = "6.0.0", optional = true }
 docs_rs_config = { path = "../docs_rs_config" }
 docs_rs_env_vars = { path = "../docs_rs_env_vars" }
 docs_rs_headers = { path = "../docs_rs_headers" }
-docs_rs_logging = { path = "../docs_rs_logging", optional = true }
 docs_rs_mimes = { path = "../docs_rs_mimes" }
 docs_rs_opentelemetry = { path = "../docs_rs_opentelemetry" }
 docs_rs_rustdoc_json = { path = "../docs_rs_rustdoc_json" }
@@ -37,6 +35,7 @@ futures-util = { workspace = true }
 http = { workspace = true }
 itertools = { workspace = true }
 mime = { workspace = true }
+moka = { version = "0.12.14", features = ["future"] }
 opentelemetry = { workspace = true }
 rand = { workspace = true, optional = true }
 serde_json = { workspace = true }
@@ -52,9 +51,9 @@ zip = { workspace = true }
 zstd = "0.13.0"
 
 [dev-dependencies]
-criterion = "0.8.0"
+criterion = { version = "0.8.0", features = ["async_tokio"] }
+dashmap = "6.0.0"
 docs_rs_config = { path = "../docs_rs_config", features = ["testing"] }
-docs_rs_logging = { path = "../docs_rs_logging", features = ["testing"] }
 docs_rs_opentelemetry = { path = "../docs_rs_opentelemetry", features = ["testing"] }
 rand = { workspace = true }
 test-case = { workspace = true }
@@ -63,5 +62,10 @@ test-case = { workspace = true }
 name = "compression"
 harness = false
 
+[[bench]]
+name = "archive_index_cache"
+harness = false
+required-features = ["testing"]
+
 [lints]
 workspace = true

crates/lib/docs_rs_storage/benches/archive_index_cache.rs

@@ -0,0 +1,138 @@
+use anyhow::Result;
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
+use docs_rs_opentelemetry::testing::TestMetrics;
+use docs_rs_storage::{StorageKind, testing::TestStorage};
+use docs_rs_types::BuildId;
+use futures_util::future::try_join_all;
+use std::{
+    path::Path,
+    sync::atomic::{AtomicI32, Ordering},
+};
+use tokio::{fs, runtime};
+
+const ARCHIVE_PATH: &str = "bench/archive.zip";
+const FILE_IN_ARCHIVE: &str = "Cargo.toml";
+
+async fn write_fixture_files(root: &Path) -> Result<()> {
+    fs::create_dir_all(root.join("src")).await?;
+    fs::write(root.join("Cargo.toml"), "[package]\nname = \"bench\"\n").await?;
+    fs::write(root.join("src/lib.rs"), "pub fn f() -> usize { 42 }\n").await?;
+    Ok(())
+}
+
+async fn create_storage_and_archive() -> Result<TestStorage> {
+    let metrics = TestMetrics::new();
+    let storage = TestStorage::from_kind(StorageKind::Memory, metrics.provider()).await?;
+
+    let fixture_dir = tempfile::tempdir()?;
+    write_fixture_files(fixture_dir.path()).await?;
+
+    storage
+        .store_all_in_archive(ARCHIVE_PATH, fixture_dir.path())
+        .await?;
+
+    Ok(storage)
+}
+
+pub fn archive_index_cache(c: &mut Criterion) {
+    let runtime = runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+
+    let storage = runtime
+        .block_on(create_storage_and_archive())
+        .expect("can't create test storage & archive");
+
+    runtime.block_on(async {
+        assert!(
+            storage
+                .exists_in_archive(ARCHIVE_PATH, Some(BuildId(1)), FILE_IN_ARCHIVE)
+                .await
+                .expect("initial exists-call failed")
+        );
+    });
+
+    let mut group = c.benchmark_group("archive_index_cache");
+
+    group.bench_function(
+        BenchmarkId::new("hot_local_index_single", "exists_in_archive"),
+        |b| {
+            b.to_async(&runtime).iter(|| async {
+                assert!(
+                    storage
+                        .exists_in_archive(ARCHIVE_PATH, Some(BuildId(1)), FILE_IN_ARCHIVE)
+                        .await
+                        .unwrap()
+                );
+            });
+        },
+    );
+
+    let cold_counter = AtomicI32::new(10_000);
+    group.bench_function(
+        BenchmarkId::new("cold_index_single", "exists_in_archive"),
+        |b| {
+            b.to_async(&runtime).iter(|| async {
+                let build_id = BuildId(cold_counter.fetch_add(1, Ordering::Relaxed));
+                assert!(
+                    storage
+                        .exists_in_archive(ARCHIVE_PATH, Some(build_id), FILE_IN_ARCHIVE)
+                        .await
+                        .unwrap()
+                );
+            });
+        },
+    );
+
+    let concurrent_counter = AtomicI32::new(20_000);
+    group.bench_function(
+        BenchmarkId::new("cold_index_concurrent_same_key_16", "exists_in_archive"),
+        |b| {
+            b.to_async(&runtime).iter(|| async {
+                let build_id = BuildId(concurrent_counter.fetch_add(1, Ordering::Relaxed));
+                let futures = (0..16).map(|_| {
+                    storage.exists_in_archive(ARCHIVE_PATH, Some(build_id), FILE_IN_ARCHIVE)
+                });
+
+                let results = try_join_all(futures).await.unwrap();
+                assert!(results.into_iter().all(std::convert::identity));
+            });
+        },
+    );
+
+    let recover_counter = AtomicI32::new(30_000);
+    group.bench_function(
+        BenchmarkId::new("purge_then_recover_single", "exists_in_archive"),
+        |b| {
+            b.to_async(&runtime).iter(|| async {
+                let build_id = BuildId(recover_counter.fetch_add(1, Ordering::Relaxed));
+                assert!(
+                    storage
+                        .exists_in_archive(ARCHIVE_PATH, Some(build_id), FILE_IN_ARCHIVE)
+                        .await
+                        .unwrap()
+                );
+
+                let local_index_path = storage
+                    .config()
+                    .archive_index_cache
+                    .path
+                    .join(format!("{ARCHIVE_PATH}.{}.index", build_id.0));
+                let _ = fs::remove_file(&local_index_path).await;
+
+                assert!(
+                    storage
+                        .exists_in_archive(ARCHIVE_PATH, Some(build_id), FILE_IN_ARCHIVE)
+                        .await
+                        .unwrap()
+                );
+            });
+        },
+    );
+
+    group.finish();
+}
+
+criterion_group!(archive_index_cache_benches, archive_index_cache);
+criterion_main!(archive_index_cache_benches);