fix(lock): fence write commit on lock loss (#4406)

fix(lock): fence write commit on lock loss (backlog#899 Phase 2)

Phase 0+1 (#4388) made object write locks refreshable and marks the guard
lost when the heartbeat can no longer refresh a quorum, but does not act on
it. Under a partition a long write's lock can expire on an unreachable node
and be reclaimed, letting a third party re-acquire it; the original writer
keeps going and both commit -- a double write.

Expose the loss signal through NamespaceLockGuard::is_lock_lost() and
ObjectLockDiagGuard::is_lock_lost(), and fence the commit in put_object and
complete_multipart_upload: immediately before rename_data (the atomic commit
point), abort with a retryable NamespaceLockQuorumUnavailable (503) if the
lock was lost. In multipart the check precedes cleanup_multipart_path so a
lost lock leaves the upload intact and retryable. A write that already
reached rename_data Ok is durable and never aborted.

The loss criterion is unchanged (reacts to Phase 1's signal). Heal and the
long-GET read side are deferred follow-ups.

crates/ecstore/src/set_disk/mod.rs

@@ -254,6 +254,13 @@ impl ObjectLockDiagGuard {
             acquired_at: Instant::now(),
         }
     }
+
+    /// Whether the underlying namespace lock's heartbeat has observed a
+    /// refresh-quorum loss (backlog#899 Phase 2). Callers fence their commit
+    /// point on this so a stale lock holder does not race a double-write.
+    fn is_lock_lost(&self) -> bool {
+        self.guard.is_lock_lost()
+    }
 }
 
 impl Drop for ObjectLockDiagGuard {

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -1342,6 +1342,22 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             );
         }
 
+        // Phase 2 (backlog#899): fence the commit on lock loss before any destructive
+        // step. If the refresh heartbeat has observed a refresh-quorum loss, another
+        // writer may have re-acquired this object's lock; proceeding would race a
+        // double-write. Abort with a retryable error *before* cleanup_multipart_path
+        // (which removes the source parts) and rename_data (which commits the object),
+        // so a lost lock leaves the upload intact and retryable.
+        if object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
+            return Err(StorageError::NamespaceLockQuorumUnavailable {
+                mode: "complete_multipart_upload_commit",
+                bucket: bucket.to_string(),
+                object: object.to_string(),
+                required: 1,
+                achieved: 0,
+            });
+        }
+
         let complete_tail_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
         self.cleanup_multipart_path(&parts).await;
 

crates/ecstore/src/set_disk/ops/object.rs

@@ -864,6 +864,22 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 object_lock_guard = Some(self.acquire_write_lock_diag("put_object_commit", bucket, object).await?);
             }
 
+            // Phase 2 (backlog#899): fence the commit on lock loss. If the refresh
+            // heartbeat has observed a refresh-quorum loss, another writer may have
+            // re-acquired this object's lock; committing now would race a double-write.
+            // Abort with a retryable error *before* rename_data makes the new version
+            // durable — once rename_data returns Ok the write is committed and must
+            // never be aborted (that would violate "a committed write is not lost").
+            if object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
+                return Err(StorageError::NamespaceLockQuorumUnavailable {
+                    mode: "put_object_commit",
+                    bucket: bucket.to_string(),
+                    object: object.to_string(),
+                    required: 1,
+                    achieved: 0,
+                });
+            }
+
             let rename_stage_start = Instant::now();
             let (online_disks, _, op_old_dir, cleanup_disks) = Self::rename_data(
                 &shuffle_disks,

crates/lock/src/distributed_lock.rs

@@ -1214,6 +1214,31 @@ mod tests {
         drop(guard);
     }
 
+    // Phase 2 passthrough: NamespaceLockGuard::Standard must forward the distributed
+    // guard's lock-loss signal so the write path can fence its commit on it.
+    #[tokio::test]
+    async fn namespace_guard_forwards_lock_lost() {
+        let (clients, _counters) = counting_clients(&[
+            RefreshOutcome::NotFound,
+            RefreshOutcome::NotFound,
+            RefreshOutcome::Alive,
+            RefreshOutcome::Alive,
+        ]);
+        let lock = DistributedLock::new("test".to_string(), clients, 3);
+        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
+            .with_ttl(Duration::from_millis(120))
+            .with_refresh_interval(Duration::from_millis(20));
+
+        let guard = lock.acquire_guard(&request).await.unwrap().unwrap();
+        tokio::time::sleep(Duration::from_millis(60)).await; // at least one tick -> lost
+
+        let ns_guard = crate::namespace::NamespaceLockGuard::Standard(guard);
+        assert!(
+            ns_guard.is_lock_lost(),
+            "NamespaceLockGuard::Standard must forward the distributed guard's lock-loss signal"
+        );
+    }
+
     // A5 -- refresh RPC jitter (Err) is not counted as not_found; the lock is not declared lost.
     #[tokio::test]
     async fn heartbeat_rpc_error_not_counted_as_lock_lost() {

crates/lock/src/namespace/mod.rs

@@ -118,6 +118,16 @@ impl NamespaceLockGuard {
             Self::Fast(guard) => guard.is_released(),
         }
     }
+
+    /// Whether the distributed guard's refresh heartbeat has observed a
+    /// refresh-quorum loss (backlog#899 Phase 2). Local (fast) locks are
+    /// single-node and never lose quorum, so they always report `false`.
+    pub fn is_lock_lost(&self) -> bool {
+        match self {
+            Self::Standard(guard) => guard.is_lock_lost(),
+            Self::Fast(_) => false,
+        }
+    }
 }
 
 /// Namespace lock for managing locks by resource namespaces