perf(locks): read bucket lock bodies concurrently and count locks for stats

server/src/locks.rs

@@ -1,6 +1,7 @@
 use std::path::{Path, PathBuf};
 use std::time::Duration;
 
+use futures_util::StreamExt;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use time::OffsetDateTime;
@@ -57,6 +58,8 @@ const DEFAULT_CAPACITY: usize = 10_000;
 
 const GATES: usize = 64;
 
+const READ_AT_ONCE: usize = 16;
+
 pub struct Held {
     pub lock: Lock,
     version: Option<String>,
@@ -343,7 +346,7 @@ impl LockStore {
     // every genuinely new create: routing it through `list` made taking one
     // lock cost a read of every lock already held. A count is a directory walk
     // or a key listing, and neither opens a single body.
-    async fn count(&self, ns: &Namespace) -> Result<usize, Error> {
+    pub async fn count(&self, ns: &Namespace) -> Result<usize, Error> {
         match &self.backend {
             Backend::Local { root } => {
                 let Ok(mut entries) = fs::read_dir(Self::directory_in(root, ns)).await else {
@@ -395,10 +398,14 @@ impl LockStore {
     // unhelpful; for locks it tells a client every file is free, which is the
     // one answer that loses somebody's work.
     async fn list_bucket(bucket: &Keyspace, ns: &Namespace) -> Result<Vec<Lock>, Error> {
-        let mut locks = Vec::new();
-
-        for key in bucket.keys(&Self::prefix(ns)).await? {
-            if let Some(bytes) = bucket.get_bytes(&key).await?
+        let keys = bucket.keys(&Self::prefix(ns)).await?;
+        let mut locks = Vec::with_capacity(keys.len());
+        let mut bodies = futures_util::stream::iter(keys)
+            .map(|key| async move { bucket.get_bytes(&key).await })
+            .buffer_unordered(READ_AT_ONCE);
+
+        while let Some(body) = bodies.next().await {
+            if let Some(bytes) = body?
                 && let Ok(lock) = serde_json::from_slice(&bytes)
             {
                 locks.push(lock);

server/src/locks/tests.rs

@@ -529,6 +529,55 @@ async fn creating_a_lock_counts_the_others_and_reads_none_of_them() {
     );
 }
 
+#[tokio::test]
+async fn a_bucket_lists_every_lock_in_path_order_past_one_batch_of_reads() {
+    crate::tls::install_crypto_provider();
+    let (endpoint, _objects) = crate::storage::s3::tests::bucket().await;
+    let locks = LockStore::bucket(crate::storage::s3::tests::keyspace(&endpoint));
+    let ns = namespace();
+    let held = READ_AT_ONCE * 2 + 3;
+    for scene in (0..held).rev() {
+        locks
+            .create(&ns, &format!("Assets/{scene:03}.unity"), &actor("jane"))
+            .await
+            .unwrap();
+    }
+
+    let paths: Vec<String> = locks
+        .list(&ns)
+        .await
+        .unwrap()
+        .into_iter()
+        .map(|lock| lock.path)
+        .collect();
+
+    let expected: Vec<String> = (0..held)
+        .map(|scene| format!("Assets/{scene:03}.unity"))
+        .collect();
+    assert_eq!(paths, expected);
+}
+
+#[tokio::test]
+async fn counting_bucket_locks_reads_no_bodies() {
+    crate::tls::install_crypto_provider();
+    let (endpoint, _objects, reads) = crate::storage::s3::tests::bucket_counting_key_reads().await;
+    let locks = LockStore::bucket(crate::storage::s3::tests::keyspace(&endpoint));
+    let ns = namespace();
+    for scene in 0..20 {
+        locks
+            .create(&ns, &format!("Assets/{scene:02}.unity"), &actor("jane"))
+            .await
+            .unwrap();
+    }
+
+    let before = reads.load(std::sync::atomic::Ordering::SeqCst);
+    let counted = locks.count(&ns).await.unwrap();
+    let cost = reads.load(std::sync::atomic::Ordering::SeqCst) - before;
+
+    assert_eq!(counted, 20);
+    assert_eq!(cost, 0, "counting 20 locks read {cost} bodies");
+}
+
 #[derive(Clone, Default)]
 struct Captured(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
 

server/src/routes.rs

@@ -142,7 +142,7 @@ async fn stats(
     Ok(Json(StatsResponse {
         objects,
         bytes,
-        locks: state.locks.list(&ns).await?.len(),
+        locks: state.locks.count(&ns).await?,
     }))
 }
 

server/tests/dashboard.rs

@@ -5,7 +5,8 @@ use std::time::Duration;
 use axum::Router;
 use axum::body::Body;
 use axum::http::{Request, StatusCode};
-use common::{app, credentials, forge, put, read_json};
+use common::{app, credentials, forge, post, put, read_json};
+use serde_json::json;
 use tower::ServiceExt;
 
 async fn get(app: Router, path: &str, token: Option<&str>) -> axum::response::Response {
@@ -34,6 +35,27 @@ async fn stats_report_what_this_repository_holds() {
     assert_eq!(body["locks"], 0);
 }
 
+#[tokio::test]
+async fn stats_count_the_locks_held() {
+    let root = tempfile::tempdir().unwrap();
+    let (api_url, _forge) = forge().await;
+    let app = app(&root, &api_url, Duration::from_secs(60));
+    for path in ["Assets/Arena.unity", "Assets/Hero.psd"] {
+        let taken = post(
+            app.clone(),
+            "writer",
+            "/FerrLabs/LFSX/locks",
+            json!({ "path": path }),
+        )
+        .await;
+        assert_eq!(taken.status(), StatusCode::CREATED);
+    }
+
+    let body = read_json(get(app, "/FerrLabs/LFSX/objects/stats", Some("reader")).await).await;
+
+    assert_eq!(body["locks"], 2);
+}
+
 #[tokio::test]
 async fn stats_are_scoped_to_the_repository_asked_about() {
     let root = tempfile::tempdir().unwrap();