fix(cluster): fence the storage scope at the lease deadline

changelog.d/cluster-lease-deadline-fence.fixed.md

@@ -0,0 +1,3 @@
+- A server paused past its S3 lease deadline no longer sends a storage write after that deadline, whichever of its tasks resumes first. See [lease restarts][lease-deadline].
+
+[lease-deadline]: ../docs/user/operations/storage-failures.md#restart-after-a-failure

changelog.d/cluster-lease-live-owner-kept.fixed.md

@@ -0,0 +1,3 @@
+- A restarted server no longer takes the S3 cluster lock from a server that is still heartbeating, and a server whose lock was taken over or unlocked stops at its next heartbeat. See [lease restarts][lease-live-owner].
+
+[lease-live-owner]: ../docs/user/operations/storage-failures.md#restart-after-a-failure

crates/omnigraph-cluster/src/lib.rs

@@ -40,6 +40,7 @@ pub use lease::{LeaseFence, LeaseFenced};
 #[cfg(any(test, feature = "test-util"))]
 pub mod lease_protocol {
     pub use crate::lease::{LeaseKeeper, Watch, lease_dir, watch_and_claim};
+    pub use crate::state_lock::leased_lock_record;
 }
 pub use admission::{
     ClusterAdmission, ClusterAdmissionPurpose, acquire_cluster_admission, acquire_graph_admission,

crates/omnigraph-cluster/src/state_lock.rs

@@ -21,6 +21,9 @@ const RELEASED_LOCK_VERSION: u32 = 2;
 /// take it over nor release it.
 const LEASED_LOCK_VERSION: u32 = 3;
 const CLUSTER_LOCK_FILE: &str = "__cluster/lock.json";
+/// The bound on every read of `lock.json`. An owner that reads with a
+/// smaller one gets a limit error and stops at its deadline.
+pub(crate) const LOCK_RECORD_MAX_BYTES: u64 = 64 * 1024;
 
 #[derive(Debug, Error)]
 pub(crate) enum StateLockError {
@@ -112,7 +115,7 @@ impl StateLockFile {
     }
 
     /// The released marker older binaries also read: version 2, no lease.
-    fn mark_released(mut self) -> Self {
+    pub(crate) fn mark_released(mut self) -> Self {
         self.version = RELEASED_LOCK_VERSION;
         self.release_id = Some(Ulid::new().to_string());
         self.lease_seconds = None;
@@ -136,6 +139,15 @@ impl StateLockFile {
     }
 }
 
+/// A held, leased serving record for `lock_id`, as an acquisition or a
+/// takeover writes it, for simulations that drive the lease directly.
+#[cfg(any(test, feature = "test-util"))]
+pub fn leased_lock_record(lock_id: &str, lease_seconds: u32) -> String {
+    let mut lock = StateLockFile::held(SERVE_OPERATION, StorageKind::S3, lease_seconds);
+    lock.lock_id = lock_id.to_owned();
+    serde_json::to_string_pretty(&lock).expect("a lock record encodes")
+}
+
 /// Result of storage-native conditional lock acquisition.
 #[derive(Debug)]
 pub(crate) enum StateLockAcquire {
@@ -186,6 +198,7 @@ impl StateLockGuard {
                     let keeper = crate::lease::LeaseKeeper::start(
                         lease_adapter,
                         &root,
+                        &self.uri,
                         self.lock.lock_id(),
                         seconds,
                         lock_sent,
@@ -325,7 +338,7 @@ async fn acquire_state_lock_leased(
         // refresh a failed CAS or adopt a write whose acknowledgement was
         // lost.
         let Some((text, version)) = adapter
-            .read_text_versioned_if_exists_bounded(lock_uri, 64 * 1024)
+            .read_text_versioned_if_exists_bounded(lock_uri, LOCK_RECORD_MAX_BYTES)
             .await?
         else {
             return Ok(StateLockAcquire::Held);
@@ -389,6 +402,10 @@ async fn acquire_state_lock_leased(
 /// the loser refuses. A replacement that does not apply leaves `lock.json` at
 /// that version, so the next start takes it over the same way. Returns when
 /// the replacement was sent, if it applied.
+///
+/// The previous owner's lease directory stays: an owner whose clock stopped
+/// through its silence resumes with its own deadline ahead, and its claim
+/// and the replaced record are what stop it at its next beat.
 #[allow(clippy::too_many_arguments)]
 async fn take_over(
     adapter: &Arc<dyn StorageAdapter>,
@@ -428,9 +445,6 @@ async fn take_over(
             successor = lock.lock_id(),
             "took over a silent owner's cluster lock"
         );
-        let _ = lease_adapter
-            .delete_prefix(&crate::lease::lease_dir(root, previous.lock_id()))
-            .await;
     }
     Ok(replaced.then_some(sent))
 }
@@ -444,7 +458,7 @@ pub(crate) async fn release_settled_lock(
     lock_id: &str,
 ) -> Result<(), StateLockError> {
     let (text, version) = adapter
-        .read_text_versioned_if_exists_bounded(uri, 64 * 1024)
+        .read_text_versioned_if_exists_bounded(uri, LOCK_RECORD_MAX_BYTES)
         .await?
         .ok_or_else(|| StateLockError::InvalidBinding("held lock is absent".into()))?;
     let lock = StateLockFile::parse(&text)?;
@@ -524,7 +538,7 @@ async fn read_bootstrap_lock(
     uri: &str,
 ) -> Result<(StateLockFile, String), StateLockError> {
     let (text, version) = adapter
-        .read_text_versioned_if_exists_bounded(uri, 64 * 1024)
+        .read_text_versioned_if_exists_bounded(uri, LOCK_RECORD_MAX_BYTES)
         .await?
         .ok_or_else(|| StateLockError::InvalidBinding("bootstrap lock is absent".into()))?;
     let lock = StateLockFile::parse(&text)?;
@@ -729,11 +743,94 @@ mod tests {
                 .list_dir(&crate::lease::lease_dir(&root, &previous))
                 .await
                 .unwrap()
-                .is_empty(),
-            "the superseded owner's beats are removed"
+                .iter()
+                .any(|uri| uri.ends_with("/superseded.json")),
+            "the superseded owner's claim stays"
         );
     }
 
+    /// An owner whose clock stopped, as in a suspended VM, misses the silence
+    /// a successor waited out: when it resumes, its own deadline is still
+    /// ahead. It must stop at its next beat all the same, before it admits a
+    /// lease's worth of work beside the successor. The takeover leaves the
+    /// claim in place, and the owner reads `lock.json` before each beat,
+    /// which never names it again, so it fences even when the lease
+    /// directory is gone, as an earlier binary's takeover left it. Each
+    /// process has a paused runtime of its own; the owner's is not driven
+    /// while the successor watches, so its clock and its beats stop.
+    #[test]
+    fn a_superseded_owner_whose_clock_stopped_fences_at_its_next_beat_issue_1067() {
+        const LEASE: u32 = 2;
+        let paused = || {
+            tokio::runtime::Builder::new_current_thread()
+                .enable_time()
+                .start_paused(true)
+                .build()
+                .unwrap()
+        };
+        for directory_removed in [false, true] {
+            let (owner_runtime, successor_runtime) = (paused(), paused());
+            let store: Arc<dyn StorageAdapter> =
+                Arc::new(omnigraph_storage::ObjectStorageAdapter::in_memory());
+            let lock_uri = "/lease-tests/__cluster/lock.json";
+            let root = cluster_root_from_lock_uri(lock_uri).unwrap();
+            let acquire = || {
+                acquire_state_lock_leased(
+                    store.clone(),
+                    store.clone(),
+                    StorageKind::S3,
+                    lock_uri,
+                    SERVE_OPERATION,
+                    LEASE,
+                )
+            };
+            let StateLockAcquire::Acquired(owner) = owner_runtime.block_on(acquire()).unwrap()
+            else {
+                panic!("a fresh root is free");
+            };
+            let StateLockAcquire::Acquired(successor) =
+                successor_runtime.block_on(acquire()).unwrap()
+            else {
+                panic!("an owner whose clock stopped is silent, and is taken over");
+            };
+            let dir = crate::lease::lease_dir(&root, owner.lock_id());
+            if directory_removed {
+                successor_runtime
+                    .block_on(store.delete_prefix(&dir))
+                    .unwrap();
+            }
+            // The owner's clock resumes where it stopped.
+            let fenced = crate::lease::LeaseFence(owner.lease_fenced()).fenced();
+            let fence = owner_runtime.block_on(async {
+                tokio::time::timeout(std::time::Duration::from_secs(u64::from(LEASE)), fenced)
+                    .await
+                    .ok()
+            });
+            assert_eq!(
+                fence,
+                Some(crate::lease::LeaseFenced::Superseded),
+                "directory removed: {directory_removed}"
+            );
+            let stored = successor_runtime
+                .block_on(store.read_text(lock_uri))
+                .unwrap();
+            assert_eq!(
+                StateLockFile::parse(&stored).unwrap().lock_id(),
+                successor.lock_id()
+            );
+            if !directory_removed {
+                assert!(
+                    successor_runtime
+                        .block_on(store.list_dir(&dir))
+                        .unwrap()
+                        .iter()
+                        .any(|uri| uri.ends_with("/superseded.json")),
+                    "the takeover keeps the claim"
+                );
+            }
+        }
+    }
+
     /// A successor that claimed and then stopped before its if-match leaves
     /// the claim and the dead owner's record. The next start still takes
     /// the lock over after its own silence, and of two starts racing, the
@@ -808,7 +905,7 @@ mod tests {
             panic!("a fresh root is free");
         };
         let (_, read_by_owner) = store
-            .read_text_versioned_if_exists_bounded(lock_uri, 64 * 1024)
+            .read_text_versioned_if_exists_bounded(lock_uri, LOCK_RECORD_MAX_BYTES)
             .await
             .unwrap()
             .unwrap();

crates/omnigraph-cluster/src/store.rs

@@ -19,7 +19,8 @@ use omnigraph_storage::{
 
 use crate::deployment::{DeploymentBundle, MAX_BUNDLE_BYTES, MAX_LEDGER_BYTES, validate_state};
 use crate::state_lock::{
-    StateLockAcquire, StateLockError, StateLockGuard, acquire_state_lock, release_settled_lock,
+    LOCK_RECORD_MAX_BYTES, StateLockAcquire, StateLockError, StateLockGuard, acquire_state_lock,
+    release_settled_lock,
 };
 use crate::{
     CLUSTER_LOCK_FILE, CLUSTER_RECOVERIES_DIR, CLUSTER_RESOURCES_DIR, CLUSTER_STATE_FILE,
@@ -548,7 +549,7 @@ impl ClusterStore {
             .read_text_versioned_if_exists_bounded(
                 uri,
                 if uri == self.uri(CLUSTER_LOCK_FILE) {
-                    64 * 1024
+                    LOCK_RECORD_MAX_BYTES
                 } else {
                     MAX_LEDGER_BYTES as u64
                 },

crates/omnigraph-dst/tests/scenarios.rs

@@ -6839,13 +6839,16 @@ async fn dst_publication_requests_get_a_definite_outcome_issue_1067() {
 /// at a seeded moment that can fall before, during or after a successor's
 /// watch. A beat that lands during the watch sends the successor away, which
 /// is the safe direction; the successor that claims the lock does so only
-/// after a lease and a half of silence since the last beat it saw; and the
-/// dead owner's beats that land after the claim fence nothing: the
-/// successor keeps renewing, and a third process watching it refuses.
+/// after a lease and a half of silence since the last beat it saw, then
+/// replaces the record by if-match on the version it read; and the dead
+/// owner's beats that land after the claim fence nothing: the successor
+/// keeps renewing, and a third process watching it refuses.
 #[test]
 #[serial]
 fn dst_successor_takes_over_a_dead_owners_lease_issue_1067() {
-    use omnigraph_cluster::lease_protocol::{LeaseKeeper, Watch, lease_dir, watch_and_claim};
+    use omnigraph_cluster::lease_protocol::{
+        LeaseKeeper, Watch, lease_dir, leased_lock_record, watch_and_claim,
+    };
     use std::time::Duration;
     use tokio::time::Instant;
 
@@ -6863,10 +6866,27 @@ fn dst_successor_takes_over_a_dead_owners_lease_issue_1067() {
             let store: Arc<dyn omnigraph_storage::StorageAdapter> =
                 Arc::new(omnigraph_storage::ObjectStorageAdapter::in_memory());
             let root = "cluster";
-            let owner =
-                LeaseKeeper::start(store.clone(), root, "owner", SECONDS, Instant::now(), None)
-                    .await
-                    .expect("the owner's first beat lands");
+            let lock_uri = format!("{root}/{}", omnigraph_cluster::CLUSTER_LOCK_FILE);
+            let lock_sent = Instant::now();
+            store
+                .write_text(&lock_uri, &leased_lock_record("owner", SECONDS))
+                .await
+                .unwrap();
+            let (_, version) = store
+                .read_text_versioned(&lock_uri)
+                .await
+                .expect("the owner's record is visible");
+            let owner = LeaseKeeper::start(
+                store.clone(),
+                root,
+                &lock_uri,
+                "owner",
+                SECONDS,
+                lock_sent,
+                None,
+            )
+            .await
+            .expect("the owner's first beat lands");
             tokio::time::sleep(Duration::from_millis(rng.below(90_000))).await;
             drop(owner);
             let died = Instant::now();
@@ -6919,12 +6939,26 @@ fn dst_successor_takes_over_a_dead_owners_lease_issue_1067() {
                 "seed {seed}"
             );
 
+            let replaced = Instant::now();
+            assert!(
+                store
+                    .write_text_if_match(
+                        &lock_uri,
+                        &leased_lock_record("successor", SECONDS),
+                        &version
+                    )
+                    .await
+                    .unwrap()
+                    .is_some(),
+                "seed {seed}: the successor's if-match applies"
+            );
             let successor = LeaseKeeper::start(
                 store.clone(),
                 root,
+                &lock_uri,
                 "successor",
                 SECONDS,
-                Instant::now(),
+                replaced,
                 None,
             )
             .await

crates/omnigraph-storage/Cargo.toml

@@ -21,6 +21,8 @@ url = { workspace = true }
 
 [dev-dependencies]
 tempfile = { workspace = true }
+# Paused time for the scope's deadline tests.
+tokio = { workspace = true, features = ["test-util"] }
 
 [lints]
 workspace = true