Make idle pantsd lease extension cheap

docs/notes/2.34.x.md

@@ -57,6 +57,8 @@ Fixed pantsd logging remote Authentication bearer tokens in plaintext when `remo
 
 Pants now logs a warning when it cannot delete a local process execution sandbox. Cleanup failures were previously ignored, so sandboxes containing read-only directories leaked in the temporary directory without any message.
 
+An idle `pantsd` no longer re-expands and rewrites the lease of every store entry its graph references every 72 seconds. A lease with more than half of its time left is no longer rewritten, and a digest leased within the last quarter of the lease time (30 minutes) is not expanded again, which removes most of the CPU time and disk writes an idle daemon spent extending leases ([#23694](https://github.com/pantsbuild/pants/issues/23694)).
+
 ### Goals
 
 ### Backends

src/rust/fs/store/src/lib.rs

@@ -1088,11 +1088,21 @@ impl Store {
         }
     }
 
+    ///
+    /// Leases `digests` and everything below them that exists locally.
+    ///
+    /// A digest this method leased within the last quarter of the lease time is skipped, along with
+    /// everything below it: those leases are still long enough to wait for a later call, and
+    /// expanding and re-leasing every digest a long-lived graph holds is most of the cost of pantsd's
+    /// periodic lease extension.
+    ///
     pub async fn lease_all_recursively<'a, Ds: Iterator<Item = &'a Digest>>(
         &self,
         digests: Ds,
     ) -> Result<(), StoreError> {
-        let reachable_digests_and_types = self.expand_local_digests(digests).await?;
+        let started = Instant::now();
+        let digests = self.local.digests_to_renew(digests.copied(), started);
+        let reachable_digests_and_types = self.expand_local_digests(digests.iter()).await?;
         // Lease all Digests which existed (ignoring any that didn't).
         self.local
             .lease_all(
@@ -1101,6 +1111,7 @@ impl Store {
                     .flat_map(|(digest, maybe_type)| maybe_type.map(|t| (digest, t))),
             )
             .await?;
+        self.local.record_renewed(&digests, started);
         Ok(())
     }
 

src/rust/fs/store/src/local.rs

@@ -363,8 +363,16 @@ impl UnderlyingByteStore for ShardedFSDB {
 
     async fn lease(&self, fingerprint: Fingerprint) -> Result<(), String> {
         let path = self.get_path(fingerprint);
+        // As in `ShardedLmdb::lease`: a lease with more than half its time left is not rewritten.
+        let renew_before = SystemTime::now() - self.lease_time / 2;
         self.executor
             .spawn_blocking(move || {
+                if std::fs::metadata(&path)
+                    .and_then(|metadata| metadata.modified())
+                    .is_ok_and(|mtime| mtime > renew_before)
+                {
+                    return Ok(());
+                }
                 fs_set_times::set_mtime(&path, fs_set_times::SystemTimeSpec::SymbolicNow)
                     .map_err(|e| format!("Failed to extend mtime of {path:?}: {e}"))
             })
@@ -558,6 +566,8 @@ struct InnerStore {
     directory_lmdb: Result<Arc<ShardedLmdb>, String>,
     file_fsdb: ShardedFSDB,
     file_source: FileSource,
+    lease_time: Duration,
+    renewed_at: Mutex<HashMap<Digest, Instant>>,
 }
 
 impl ByteStore {
@@ -615,6 +625,8 @@ impl ByteStore {
                 file_source: FileSource {
                     open_files: Semaphore::new(1024),
                 },
+                lease_time: options.lease_time,
+                renewed_at: Mutex::default(),
             }),
         })
     }
@@ -650,6 +662,34 @@ impl ByteStore {
         }
     }
 
+    ///
+    /// The digests among `digests` that were not passed to `record_renewed` within the last
+    /// quarter of the lease time.
+    ///
+    /// `lease_all` leaves every lease it is given with at least half of the lease time left (see
+    /// `ShardedLmdb::lease`), so a digest renewed more recently than that, and everything below it,
+    /// still has at least a quarter of the lease time left: enough to wait for a later pass.
+    ///
+    pub(crate) fn digests_to_renew(
+        &self,
+        digests: impl Iterator<Item = Digest>,
+        now: Instant,
+    ) -> Vec<Digest> {
+        let renew_after = self.inner.lease_time / 4;
+        let mut renewed_at = self.inner.renewed_at.lock();
+        renewed_at.retain(|_, at| now.duration_since(*at) < renew_after);
+        digests
+            .filter(|digest| !renewed_at.contains_key(digest))
+            .collect()
+    }
+
+    pub(crate) fn record_renewed(&self, digests: &[Digest], at: Instant) {
+        self.inner
+            .renewed_at
+            .lock()
+            .extend(digests.iter().map(|digest| (*digest, at)));
+    }
+
     pub async fn lease_all(
         &self,
         digests: impl Iterator<Item = (Digest, EntryType)>,

src/rust/fs/store/src/local_tests.rs

@@ -6,7 +6,7 @@ use crate::{EntryType, LocalOptions, ShrinkBehavior};
 use std::collections::HashSet;
 use std::io::Write;
 use std::path::Path;
-use std::time::Duration;
+use std::time::{Duration, SystemTime};
 
 use bytes::{BufMut, Bytes, BytesMut};
 use hashing::{Digest, Fingerprint};
@@ -750,6 +750,40 @@ async fn get_missing_digests() {
     )
 }
 
+async fn large_file_mtime_after_lease(file_age: Duration) -> (SystemTime, SystemTime) {
+    let dir = TempDir::new().unwrap();
+    let store = new_store(dir.path());
+    let large_testdata = TestData::new("123456789".repeat(1000 * 512).as_str());
+    let digest = prime_store_with_file_bytes(&store, large_testdata.bytes()).await;
+    let path = WalkDir::new(dir.path())
+        .into_iter()
+        .map(|entry| entry.unwrap().into_path())
+        .find(|path| path.file_name() == Some(digest.hash.to_hex().as_ref()))
+        .expect("large file should be stored in the fsdb");
+    let aged_mtime = SystemTime::now() - file_age;
+    fs_set_times::set_mtime(&path, fs_set_times::SystemTimeSpec::Absolute(aged_mtime)).unwrap();
+
+    store
+        .lease_all(vec![(digest, EntryType::File)].into_iter())
+        .await
+        .expect("Error leasing");
+
+    let mtime = std::fs::metadata(&path).unwrap().modified().unwrap();
+    (aged_mtime, mtime)
+}
+
+#[tokio::test]
+async fn lease_renews_large_file_with_less_than_half_left() {
+    let (aged_mtime, mtime) = large_file_mtime_after_lease(Duration::from_secs(90 * 60)).await;
+    assert!(mtime > aged_mtime);
+}
+
+#[tokio::test]
+async fn lease_keeps_large_file_with_more_than_half_left() {
+    let (aged_mtime, mtime) = large_file_mtime_after_lease(Duration::from_secs(30 * 60)).await;
+    assert_eq!(mtime, aged_mtime);
+}
+
 pub fn new_store<P: AsRef<Path>>(dir: P) -> ByteStore {
     ByteStore::new(task_executor::Executor::new(), dir).unwrap()
 }

src/rust/fs/store/src/tests.rs

@@ -4,7 +4,7 @@ use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 use std::io::Read;
 use std::os::unix::fs::PermissionsExt;
 use std::path::{Path, PathBuf};
-use std::time::Duration;
+use std::time::{Duration, SystemTime};
 use tempfile::TempDir;
 use testutil::data::{TestData, TestDirectory};
 
@@ -22,8 +22,8 @@ use workunit_store::WorkunitStore;
 
 use crate::local::ByteStore;
 use crate::{
-    EntryType, FileContent, MEGABYTES, RemoteProvider, RemoteStoreOptions, Snapshot, Store,
-    StoreError, StoreFileByDigest, UploadSummary,
+    EntryType, FileContent, LocalOptions, MEGABYTES, RemoteProvider, RemoteStoreOptions, Snapshot,
+    Store, StoreError, StoreFileByDigest, UploadSummary,
 };
 
 pub(crate) const STORE_BATCH_API_SIZE_LIMIT: usize = 4 * 1024 * 1024;
@@ -1843,3 +1843,64 @@ async fn big_file_mutable_root_dir() {
         assert_path(&materialize_dir.path().join("nested").join("file"));
     }
 }
+
+async fn store_large_file(store: &Store, dir: &Path) -> (Digest, PathBuf) {
+    let large_testdata = TestData::new("123456789".repeat(1000 * 512).as_str());
+    let digest = store
+        .store_file_bytes(large_testdata.bytes(), true)
+        .await
+        .unwrap();
+    let path = walkdir::WalkDir::new(dir)
+        .into_iter()
+        .map(|entry| entry.unwrap().into_path())
+        .find(|path| path.file_name() == Some(digest.hash.to_hex().as_ref()))
+        .expect("large file should be stored in the fsdb");
+    (digest, path)
+}
+
+fn set_mtime(path: &Path, mtime: SystemTime) {
+    fs_set_times::set_mtime(path, fs_set_times::SystemTimeSpec::Absolute(mtime)).unwrap();
+}
+
+fn mtime(path: &Path) -> SystemTime {
+    std::fs::metadata(path).unwrap().modified().unwrap()
+}
+
+#[tokio::test]
+async fn lease_all_recursively_skips_recently_renewed_digest() {
+    let dir = TempDir::new().unwrap();
+    let store = new_local_store(dir.path());
+    let (digest, path) = store_large_file(&store, dir.path()).await;
+    store.lease_all_recursively([digest].iter()).await.unwrap();
+
+    let aged_mtime = SystemTime::now() - Duration::from_secs(90 * 60);
+    set_mtime(&path, aged_mtime);
+    store.lease_all_recursively([digest].iter()).await.unwrap();
+
+    assert_eq!(mtime(&path), aged_mtime);
+}
+
+#[tokio::test]
+async fn lease_all_recursively_renews_digest_after_a_quarter_of_the_lease_time() {
+    let dir = TempDir::new().unwrap();
+    let lease_time = Duration::from_secs(4);
+    let store = Store::local_only_with_options(
+        task_executor::Executor::new(),
+        dir.path(),
+        dir.path(),
+        LocalOptions {
+            lease_time,
+            ..LocalOptions::default()
+        },
+    )
+    .unwrap();
+    let (digest, path) = store_large_file(&store, dir.path()).await;
+    store.lease_all_recursively([digest].iter()).await.unwrap();
+
+    let aged_mtime = SystemTime::now() - lease_time;
+    set_mtime(&path, aged_mtime);
+    tokio::time::sleep(lease_time / 4 + Duration::from_millis(100)).await;
+    store.lease_all_recursively([digest].iter()).await.unwrap();
+
+    assert!(mtime(&path) > aged_mtime);
+}

src/rust/sharded_lmdb/src/lib.rs

@@ -30,6 +30,12 @@ pub const DEFAULT_LEASE_TIME: Duration = Duration::from_secs(2 * 60 * 60);
 
 const VERSIONED_FINGERPRINT_SIZE: usize = FINGERPRINT_SIZE + 1;
 
+fn lease_from_bytes(bytes: &[u8]) -> u64 {
+    let mut array = [0_u8; 8];
+    array.copy_from_slice(bytes);
+    u64::from_le_bytes(array)
+}
+
 /// VersionedFingerprint is a byte buffer one longer than the number of bytes stored in a
 /// Fingerprint. It is just the byte pattern of a Fingerprint with the version number concatenated
 /// onto the end of it.
@@ -395,11 +401,7 @@ impl ShardedLmdb {
                         // when we delete from lmdb to track how much we've freed).
                         let lease_until_unix_timestamp = txn
                             .get(*lease_database, &key)
-                            .map(|b| {
-                                let mut array = [0_u8; 8];
-                                array.copy_from_slice(b);
-                                u64::from_le_bytes(array)
-                            })
+                            .map(lease_from_bytes)
                             .unwrap_or_else(|e| match e {
                                 lmdb::Error::NotFound => 0,
                                 e => panic!("Error reading lease, probable lmdb corruption: {e:?}"),
@@ -613,17 +615,36 @@ impl ShardedLmdb {
             .flatten()
     }
 
+    ///
+    /// Extends the lease on `fingerprint` to `lease_time` from now.
+    ///
+    /// A lease with more than half of `lease_time` left is not rewritten. pantsd renews the leases
+    /// of everything its graph references more often than that, so the entry is still renewed
+    /// before it could expire, and an idle pantsd does not rewrite every lease it holds on every
+    /// pass.
+    ///
     pub async fn lease(&self, fingerprint: Fingerprint) -> Result<(), String> {
         let store = self.clone();
         self.executor
             .spawn_blocking(move || {
                 let until_secs_since_epoch: u64 = store.lease_until_secs_since_epoch();
                 let (env, _, lease_database) = store.get(&fingerprint);
-                env.begin_rw_txn()
-                    .and_then(|mut txn| {
+                let key = VersionedFingerprint::new(fingerprint, ShardedLmdb::SCHEMA_VERSION);
+                store
+                    .lease_outlasts(
+                        &env,
+                        lease_database,
+                        &key,
+                        until_secs_since_epoch - store.lease_time.as_secs() / 2,
+                    )
+                    .and_then(|fresh| {
+                        if fresh {
+                            return Ok(());
+                        }
+                        let mut txn = env.begin_rw_txn()?;
                         store.lease_inner(
                             lease_database,
-                            &VersionedFingerprint::new(fingerprint, ShardedLmdb::SCHEMA_VERSION),
+                            &key,
                             until_secs_since_epoch,
                             &mut txn,
                         )?;
@@ -636,6 +657,21 @@ impl ShardedLmdb {
             .flatten()
     }
 
+    fn lease_outlasts(
+        &self,
+        env: &Environment,
+        lease_database: Database,
+        versioned_fingerprint: &VersionedFingerprint,
+        secs_since_epoch: u64,
+    ) -> Result<bool, lmdb::Error> {
+        let txn = env.begin_ro_txn()?;
+        match txn.get(lease_database, &versioned_fingerprint.as_ref()) {
+            Ok(bytes) => Ok(lease_from_bytes(bytes) > secs_since_epoch),
+            Err(lmdb::Error::NotFound) => Ok(false),
+            Err(e) => Err(e),
+        }
+    }
+
     fn lease_inner(
         &self,
         database: Database,