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/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