// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use crate::{
    ObjectKey,
    client::LockClient,
    error::{LockError, Result},
    types::{LockId, LockInfo, LockRequest, LockResponse, LockStatus, LockType},
};
use futures::{
    future::join_all,
    stream::{FuturesUnordered, StreamExt},
};
use rand::RngExt as _;
use rustfs_io_metrics::{
    record_lock_refresh_quorum_lost, record_read_lock_held_acquire, record_read_lock_held_release,
    record_write_lock_held_acquire, record_write_lock_held_release,
};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::Notify;
use tokio::task::{JoinHandle, JoinSet};
use tokio::time::Instant;
use tracing::{debug, warn};
use uuid::Uuid;

const UNLOCK_RETRY_ATTEMPTS: usize = 3;
const UNLOCK_RETRY_BACKOFF: Duration = Duration::from_millis(100);
const LOCK_ACQUIRE_RETRY_INITIAL_BACKOFF: Duration = Duration::from_millis(250);
const LOCK_ACQUIRE_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(1);
const REMOTE_LOCK_RPC_FAILED_PREFIX: &str = "remote lock rpc failed:";
const REMOTE_LOCK_RPC_TIMED_OUT_PREFIX: &str = "remote lock rpc timed out:";
const UNRECOVERABLE_QUORUM_FAILURE_PREFIX: &str = "unrecoverable quorum failure";

#[derive(Clone, Copy, Debug)]
enum LockAcquireFailureKind {
    NonRetryable,
    RetryableContention,
    UnrecoverableQuorum,
}

/// Generate a new aggregate lock ID for multiple client locks
fn generate_aggregate_lock_id(resource: &ObjectKey) -> LockId {
    LockId {
        resource: resource.clone(),
        uuid: Uuid::new_v4().to_string(),
    }
}

fn is_remote_lock_rpc_failure(error: &str) -> bool {
    has_case_insensitive_prefix(error, REMOTE_LOCK_RPC_FAILED_PREFIX)
        || has_case_insensitive_prefix(error, REMOTE_LOCK_RPC_TIMED_OUT_PREFIX)
}

fn is_remote_lock_rpc_timeout(error: &str) -> bool {
    has_case_insensitive_prefix(error, REMOTE_LOCK_RPC_TIMED_OUT_PREFIX)
}

fn has_case_insensitive_prefix(error: &str, expected_prefix: &str) -> bool {
    error
        .get(0..expected_prefix.len())
        .is_some_and(|prefix| prefix.eq_ignore_ascii_case(expected_prefix))
}

fn is_unrecoverable_quorum_error(error: &str) -> bool {
    error
        .get(0..UNRECOVERABLE_QUORUM_FAILURE_PREFIX.len())
        .is_some_and(|prefix| prefix.eq_ignore_ascii_case(UNRECOVERABLE_QUORUM_FAILURE_PREFIX))
}

/// Derive a safe heartbeat interval for a freshly acquired guard.
///
/// Deliberately avoids `Duration::clamp` (which asserts `min <= max` and would panic for
/// sub-second ttls where `ttl - 1s` underflows to zero). Returns `None` for degenerate cases so
/// no heartbeat is spawned:
/// - `entries_len == 0`: no acquired backend lease to renew,
/// - `interval.is_zero()` or `interval >= ttl`: too small a ttl / too large an interval to renew.
fn derive_refresh_interval(entries_len: usize, ttl: Duration, injected: Option<Duration>) -> Option<Duration> {
    if entries_len == 0 {
        return None;
    }
    let interval = injected.unwrap_or(ttl / 3);
    if interval.is_zero() || interval >= ttl {
        return None;
    }
    Some(interval)
}

fn classify_lock_failure(error: &str) -> LockAcquireFailureKind {
    if is_unrecoverable_quorum_error(error) {
        return LockAcquireFailureKind::UnrecoverableQuorum;
    }

    if error.to_ascii_lowercase().contains("timeout") || is_remote_lock_rpc_failure(error) {
        return LockAcquireFailureKind::RetryableContention;
    }

    LockAcquireFailureKind::NonRetryable
}

fn classify_lock_error_failure(error: &LockError) -> LockAcquireFailureKind {
    if error.is_retryable() {
        LockAcquireFailureKind::RetryableContention
    } else {
        classify_lock_failure(&error.to_string())
    }
}

fn should_warn_lock_failure(error: &str) -> bool {
    if is_remote_lock_rpc_failure(error) {
        return !is_remote_lock_rpc_timeout(error);
    }

    matches!(
        classify_lock_failure(error),
        LockAcquireFailureKind::NonRetryable | LockAcquireFailureKind::UnrecoverableQuorum
    )
}

/// Observability channel for lock-loss: set once when a guard's heartbeat detects it has
/// lost refresh quorum. Phase 1 only signals (warn + metric + this flag); Phase 2 (backlog#899)
/// will `select!` on `notified()` inside long write operations to abort and avoid split-brain.
#[derive(Debug, Default)]
pub struct LockLostSignal {
    lost: AtomicBool,
    valid_until: Mutex<Option<Instant>>,
    notify: Notify,
}

impl LockLostSignal {
    fn set_valid_until(&self, valid_until: Instant) {
        if self.lost.load(Ordering::SeqCst) {
            return;
        }
        match self.valid_until.lock() {
            Ok(mut deadline) => {
                if deadline.is_some_and(|current| Instant::now() >= current) {
                    drop(deadline);
                    self.mark_lost();
                    return;
                }
                *deadline = Some(valid_until);
                self.notify.notify_waiters();
            }
            Err(_) => self.mark_lost(),
        }
    }

    fn mark_lost(&self) {
        self.lost.store(true, Ordering::SeqCst);
        self.notify.notify_waiters();
    }

    /// Whether refresh quorum has been lost for the associated guard.
    pub fn is_lost(&self) -> bool {
        if self.lost.load(Ordering::SeqCst) {
            return true;
        }
        let expired = match self.valid_until.lock() {
            Ok(deadline) => deadline.is_some_and(|valid_until| Instant::now() >= valid_until),
            Err(_) => true,
        };
        if expired {
            self.mark_lost();
        }
        expired
    }

    /// Resolves once the lock is declared lost (immediately if already lost).
    pub async fn notified(&self) {
        self.notified_after_registration(|| {}).await;
    }

    async fn notified_after_registration(&self, after_registration: impl FnOnce()) {
        let mut after_registration = Some(after_registration);
        loop {
            let notified = self.notify.notified();
            tokio::pin!(notified);
            notified.as_mut().enable();
            if let Some(after_registration) = after_registration.take() {
                after_registration();
            }
            if self.is_lost() {
                return;
            }

            let valid_until = match self.valid_until.lock() {
                Ok(deadline) => *deadline,
                Err(_) => return,
            };
            match valid_until {
                Some(valid_until) => {
                    tokio::select! {
                        _ = tokio::time::sleep_until(valid_until) => {}
                        _ = &mut notified => {}
                    }
                }
                None => notified.await,
            }
        }
    }
}

#[derive(Clone, Debug)]
struct HeldLockEntry {
    lock_id: LockId,
    client: Arc<dyn LockClient>,
    valid_until: Instant,
}

impl HeldLockEntry {
    fn release_entry(&self) -> (LockId, Arc<dyn LockClient>) {
        (self.lock_id.clone(), self.client.clone())
    }
}

#[derive(Clone, Debug)]
struct LockHeartbeatConfig {
    ttl: Duration,
    interval: Option<Duration>,
    refresh_quorum: usize,
    owner: String,
    resource: ObjectKey,
}

#[derive(Clone, Copy, Debug, Default)]
struct LockRefreshStats {
    refreshed: usize,
    not_found: usize,
    refresh_errors: usize,
}

/// A RAII guard for distributed locks that releases the lock asynchronously when dropped.
#[derive(Debug)]
pub struct DistributedLockGuard {
    /// The public-facing lock id. For multi-client scenarios this is typically
    /// an aggregate id; for single-client it is the only id.
    lock_id: LockId,
    /// All underlying leases that should be refreshed and released with the guard.
    entries: Vec<HeldLockEntry>,
    lock_type: LockType,
    /// If true, Drop will not try to release (used if user manually released).
    disarmed: bool,
    /// Background lease task for renewable leases.
    refresh_task: Option<JoinHandle<()>>,
    /// Lock-loss signal, shared with the heartbeat task.
    lock_lost: Arc<LockLostSignal>,
}

impl DistributedLockGuard {
    /// Create a new guard.
    ///
    /// - `lock_id` is the id returned to the caller (`lock_id()`).
    /// - `entries` is the full list of underlying leases that should be refreshed and released
    ///   with this guard.
    /// - `heartbeat`: lease TTL, optional renewal cadence, quorum, and diagnostics.
    fn new(lock_id: LockId, entries: Vec<HeldLockEntry>, lock_type: LockType, heartbeat: LockHeartbeatConfig) -> Self {
        record_lock_held_acquire(lock_type);
        let lock_lost = Arc::new(LockLostSignal::default());
        match Self::quorum_valid_until(&entries, heartbeat.refresh_quorum) {
            Some(valid_until) => lock_lost.set_valid_until(valid_until),
            None => lock_lost.mark_lost(),
        }
        // Non-renewable guards are fenced directly by LockLostSignal's deadline checks.
        let refresh_task = heartbeat.interval.and_then(|_| {
            tokio::runtime::Handle::try_current().ok().map(|handle| {
                let entries = entries.clone();
                let lock_lost = lock_lost.clone();
                handle.spawn(Self::run_heartbeat(entries, heartbeat, lock_lost))
            })
        });
        Self {
            lock_id,
            entries,
            lock_type,
            disarmed: false,
            refresh_task,
            lock_lost,
        }
    }

    fn quorum_valid_until(entries: &[HeldLockEntry], refresh_quorum: usize) -> Option<Instant> {
        if refresh_quorum == 0 || entries.len() < refresh_quorum {
            return None;
        }

        let mut deadlines: Vec<_> = entries.iter().map(|entry| entry.valid_until).collect();
        deadlines.sort_unstable();
        deadlines.get(deadlines.len() - refresh_quorum).copied()
    }

    fn signal_refresh_quorum_lost(
        lock_lost: &LockLostSignal,
        config: &LockHeartbeatConfig,
        entries: usize,
        stats: LockRefreshStats,
    ) {
        warn!(
            resource = %config.resource,
            owner = config.owner,
            refreshed = stats.refreshed,
            not_found = stats.not_found,
            refresh_errors = stats.refresh_errors,
            entries,
            refresh_quorum = config.refresh_quorum,
            "lock refresh lost quorum"
        );
        record_lock_refresh_quorum_lost();
        lock_lost.mark_lost();
    }

    /// Refresh every tracked lease while a quorum is still known to be valid.
    ///
    /// A successful refresh extends only that entry's conservative local deadline. RPC errors
    /// retain the previous deadline, so transient jitter is tolerated but an unconfirmed lease
    /// can never remain valid beyond its backend TTL. Waiting for a slow refresh is also fenced
    /// by the current quorum deadline.
    async fn run_heartbeat(mut entries: Vec<HeldLockEntry>, config: LockHeartbeatConfig, lock_lost: Arc<LockLostSignal>) {
        let Some(interval) = config.interval else {
            return;
        };

        let now = Instant::now();
        let first_refresh = now.checked_add(interval).unwrap_or(now);
        let mut ticker = tokio::time::interval_at(first_refresh, interval);
        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

        loop {
            let Some(valid_until) = Self::quorum_valid_until(&entries, config.refresh_quorum) else {
                Self::signal_refresh_quorum_lost(&lock_lost, &config, entries.len(), LockRefreshStats::default());
                return;
            };
            lock_lost.set_valid_until(valid_until);

            tokio::select! {
                biased;
                _ = tokio::time::sleep_until(valid_until) => {
                    Self::signal_refresh_quorum_lost(&lock_lost, &config, entries.len(), LockRefreshStats::default());
                    return;
                }
                _ = ticker.tick() => {}
            }

            let mut stats = LockRefreshStats::default();
            let mut pending = FuturesUnordered::new();
            for (idx, entry) in entries.iter().enumerate() {
                let refresh_started = Instant::now();
                let lock_id = entry.lock_id.clone();
                let client = entry.client.clone();
                pending.push(async move { (idx, refresh_started, client.refresh(&lock_id).await) });
            }

            while !pending.is_empty() {
                let Some(valid_until) = Self::quorum_valid_until(&entries, config.refresh_quorum) else {
                    Self::signal_refresh_quorum_lost(&lock_lost, &config, entries.len(), stats);
                    return;
                };
                lock_lost.set_valid_until(valid_until);

                let next = tokio::select! {
                    biased;
                    _ = tokio::time::sleep_until(valid_until) => {
                        Self::signal_refresh_quorum_lost(&lock_lost, &config, entries.len(), stats);
                        return;
                    }
                    next = pending.next() => next
                };
                let Some((idx, refresh_started, result)) = next else {
                    break;
                };

                match result {
                    Ok(true) => {
                        stats.refreshed += 1;
                        entries[idx].valid_until = refresh_started.checked_add(config.ttl).unwrap_or(refresh_started);
                    }
                    Ok(false) => {
                        stats.not_found += 1;
                        entries[idx].valid_until = Instant::now();
                    }
                    Err(_) => stats.refresh_errors += 1,
                }
            }
        }
    }

    /// Get the lock id associated with this guard
    pub fn lock_id(&self) -> &LockId {
        &self.lock_id
    }

    /// Whether the guard's heartbeat has observed a refresh-quorum loss.
    pub fn is_lock_lost(&self) -> bool {
        self.lock_lost.is_lost()
    }

    /// Shared lock-loss signal (used by callers to `select!` on `notified()`; Phase 2).
    pub fn lock_lost(&self) -> Arc<LockLostSignal> {
        self.lock_lost.clone()
    }

    /// Abort the heartbeat task, if running. Idempotent.
    ///
    /// Aborting may truncate an in-flight refresh, but refresh is idempotent and only extends
    /// an existing lease (never creates one): a late refresh reaching a backend that already
    /// dropped the entry returns `Ok(false)` with no side effect.
    fn stop_heartbeat(&mut self) {
        if let Some(task) = self.refresh_task.take() {
            task.abort();
        }
    }

    /// Manually disarm the guard so dropping it won't release the lock.
    /// Call this if you explicitly released the lock elsewhere.
    pub fn disarm(&mut self) {
        // Stop renewing a lock the caller says it released elsewhere.
        self.stop_heartbeat();
        if !self.disarmed {
            record_lock_held_release(self.lock_type);
        }
        self.disarmed = true;
    }

    /// Check if the guard has been disarmed (lock already released)
    pub fn is_disarmed(&self) -> bool {
        self.disarmed
    }

    /// Manually release the lock early.
    /// This spawns a background release task and then disarms the guard
    /// to prevent double-release on drop.
    /// Returns true if release was scheduled or the guard was already disarmed.
    pub fn release(&mut self) -> bool {
        // Stop the heartbeat before releasing so no refresh races the unlock. Abort is safe even
        // if disarmed (take() makes it idempotent / a no-op when never spawned).
        self.stop_heartbeat();

        if self.disarmed {
            // Lock was already released, return true to indicate lock is in released state
            return true;
        }

        let entries = self.entries.iter().map(HeldLockEntry::release_entry).collect();
        DistributedLock::spawn_release_cleanup(entries, "distributed_lock_guard_release");

        // Disarm to prevent double-release on drop
        self.disarmed = true;
        record_lock_held_release(self.lock_type);
        true
    }
}

impl Drop for DistributedLockGuard {
    fn drop(&mut self) {
        // Call release() to handle the actual release logic
        // If already disarmed, release() will return early
        // Setting disarmed in release() is harmless here since we're dropping anyway
        let _ = self.release();
    }
}

/// Distributed lock handler for distributed use cases
/// Uses quorum-based acquisition and aggregate lock ID mapping
#[derive(Debug)]
pub struct DistributedLock {
    /// Lock clients for this namespace
    clients: Vec<Arc<dyn LockClient>>,
    /// Namespace identifier
    namespace: String,
    /// Quorum size for exclusive/write operations
    quorum: usize,
}

type LockAcquireTaskResult = (usize, Result<LockResponse>);

struct LockAcquireQuorumResult {
    response: LockResponse,
    individual_locks: Vec<HeldLockEntry>,
    failure_kind: Option<LockAcquireFailureKind>,
    quorum_impossible: bool,
}

impl DistributedLock {
    /// Create new distributed lock
    pub fn new(namespace: String, clients: Vec<Arc<dyn LockClient>>, quorum: usize) -> Self {
        let q = if clients.len() <= 1 {
            1
        } else {
            quorum.clamp(1, clients.len())
        };

        Self {
            clients,
            namespace,
            quorum: q,
        }
    }

    /// Get namespace identifier
    pub fn namespace(&self) -> &str {
        &self.namespace
    }

    fn read_quorum(&self) -> usize {
        let client_count = self.clients.len();
        if client_count <= 1 {
            1
        } else {
            client_count - (client_count / 2)
        }
    }

    fn required_quorum(&self, lock_type: LockType) -> usize {
        match lock_type {
            LockType::Shared => self.read_quorum(),
            LockType::Exclusive => self.quorum,
        }
    }

    /// Get resource key for this namespace
    pub fn get_resource_key(&self, resource: &ObjectKey) -> String {
        format!("{}:{}", self.namespace, resource)
    }

    /// Get clients (for health check and stats)
    pub(crate) fn clients(&self) -> &[Arc<dyn LockClient>] {
        &self.clients
    }

    /// Acquire a lock and return a RAII guard
    pub(crate) async fn acquire_guard(&self, request: &LockRequest) -> Result<Option<DistributedLockGuard>> {
        if self.clients.is_empty() {
            return Err(LockError::internal("No lock clients available"));
        }

        let required_quorum = self.required_quorum(request.lock_type);
        let LockAcquireQuorumResult {
            response: resp,
            individual_locks,
            failure_kind,
            ..
        } = self.acquire_lock_quorum_with_retry(request).await?;
        if resp.success {
            // Use aggregate lock_id from LockResponse's LockInfo
            // The aggregate id is what we expose to callers; individual_locks carries
            // the real (LockId, client) pairs that must be released.
            let aggregate_lock_id = resp
                .lock_info
                .as_ref()
                .map(|info| info.id.clone())
                .unwrap_or_else(|| LockId::new_unique(&request.resource));

            // Heartbeat operates on the per-client individual locks (their per-client ids), never
            // the aggregate id, so refreshes round-trip to the exact backend entries.
            let heartbeat = LockHeartbeatConfig {
                ttl: request.ttl,
                interval: derive_refresh_interval(individual_locks.len(), request.ttl, request.refresh_interval),
                refresh_quorum: required_quorum,
                owner: request.owner.clone(),
                resource: request.resource.clone(),
            };
            Ok(Some(DistributedLockGuard::new(
                aggregate_lock_id,
                individual_locks,
                request.lock_type,
                heartbeat,
            )))
        } else {
            // Check if it's a timeout or quorum failure
            if let Some(error_msg) = &resp.error {
                if request.suppress_contention_logs {
                    tracing::debug!(
                        resource = %request.resource,
                        owner = %request.owner,
                        "acquire_lock_quorum contention: {}",
                        error_msg
                    );
                } else if matches!(failure_kind, Some(LockAcquireFailureKind::NonRetryable)) {
                    warn!(
                        resource = %request.resource,
                        owner = %request.owner,
                        "acquire_lock_quorum error: {}",
                        error_msg
                    );
                } else if matches!(failure_kind, Some(LockAcquireFailureKind::RetryableContention)) {
                    debug!(
                        resource = %request.resource,
                        owner = %request.owner,
                        "acquire_lock_quorum contention: {}",
                        error_msg
                    );
                } else {
                    debug!(
                        resource = %request.resource,
                        owner = %request.owner,
                        "acquire_lock_quorum final failure: {}",
                        error_msg
                    );
                }

                if matches!(failure_kind, Some(LockAcquireFailureKind::UnrecoverableQuorum)) {
                    let achieved = individual_locks.len();
                    Err(LockError::QuorumNotReached {
                        required: required_quorum,
                        achieved,
                    })
                } else {
                    Ok(None)
                }
            } else {
                Ok(None)
            }
        }
    }

    /// Convenience: acquire exclusive lock as a guard
    pub async fn lock_guard(
        &self,
        resource: ObjectKey,
        owner: &str,
        timeout: Duration,
        ttl: Duration,
    ) -> Result<Option<DistributedLockGuard>> {
        let req = LockRequest::new(resource, LockType::Exclusive, owner)
            .with_acquire_timeout(timeout)
            .with_ttl(ttl);
        self.acquire_guard(&req).await
    }

    /// Convenience: acquire exclusive lock with expected contention logs suppressed
    pub async fn lock_guard_quiet(
        &self,
        resource: ObjectKey,
        owner: &str,
        timeout: Duration,
        ttl: Duration,
    ) -> Result<Option<DistributedLockGuard>> {
        let req = LockRequest::new(resource, LockType::Exclusive, owner)
            .with_acquire_timeout(timeout)
            .with_ttl(ttl)
            .with_suppress_contention_logs(true);
        self.acquire_guard(&req).await
    }

    /// Convenience: acquire shared lock as a guard
    pub async fn rlock_guard(
        &self,
        resource: ObjectKey,
        owner: &str,
        timeout: Duration,
        ttl: Duration,
    ) -> Result<Option<DistributedLockGuard>> {
        let req = LockRequest::new(resource, LockType::Shared, owner)
            .with_acquire_timeout(timeout)
            .with_ttl(ttl);
        self.acquire_guard(&req).await
    }

    fn spawn_lock_requests(&self, request: &LockRequest) -> JoinSet<LockAcquireTaskResult> {
        let mut pending = JoinSet::new();
        for (idx, client) in self.clients.iter().cloned().enumerate() {
            let request = request.clone();
            pending.spawn(async move { (idx, client.acquire_lock(&request).await) });
        }
        pending
    }

    fn lock_acquire_retry_backoff(attempt: usize, rng: &mut impl rand::Rng) -> Duration {
        let base = LOCK_ACQUIRE_RETRY_INITIAL_BACKOFF.saturating_mul(attempt.try_into().unwrap_or(u32::MAX));
        let jitter = base / 4;
        let lower = base.saturating_sub(jitter);
        let upper = base.saturating_add(jitter);
        let lower_nanos = u64::try_from(lower.as_nanos()).unwrap_or(u64::MAX);
        let upper_nanos = u64::try_from(upper.as_nanos()).unwrap_or(u64::MAX);

        Duration::from_nanos(rng.random_range(lower_nanos..=upper_nanos))
    }

    fn lock_acquire_attempt_timeout(&self, remaining: Duration) -> Duration {
        if self.clients.len() <= 1 {
            remaining
        } else {
            remaining.min(LOCK_ACQUIRE_ATTEMPT_TIMEOUT)
        }
    }

    async fn release_entries(entries: Vec<(LockId, Arc<dyn LockClient>)>, context: &'static str) {
        let mut pending = entries;

        for attempt in 1..=UNLOCK_RETRY_ATTEMPTS {
            let release_results = join_all(pending.into_iter().map(|(lock_id, client)| async move {
                match client.release(&lock_id).await {
                    Ok(true) => None,
                    Ok(false) => {
                        warn!(%lock_id, attempt, context, "distributed unlock did not find lock on client");
                        Some((lock_id, client))
                    }
                    Err(err) => {
                        warn!(%lock_id, attempt, context, "distributed unlock failed on client: {}", err);
                        Some((lock_id, client))
                    }
                }
            }))
            .await;

            pending = release_results.into_iter().flatten().collect();
            if pending.is_empty() {
                debug!(attempt, context, "distributed unlock completed");
                return;
            }

            if attempt < UNLOCK_RETRY_ATTEMPTS {
                tokio::time::sleep(UNLOCK_RETRY_BACKOFF * attempt as u32).await;
            }
        }

        warn!(
            remaining = pending.len(),
            attempts = UNLOCK_RETRY_ATTEMPTS,
            context,
            "distributed unlock left unreleased entries after retry"
        );
    }

    fn spawn_release_cleanup(entries: Vec<(LockId, Arc<dyn LockClient>)>, context: &'static str) {
        if entries.is_empty() {
            return;
        }

        if let Ok(handle) = tokio::runtime::Handle::try_current() {
            let join_handle = handle.spawn(async move {
                Self::release_entries(entries, context).await;
            });
            drop(join_handle);
            return;
        }

        let join_handle = std::thread::spawn(move || match tokio::runtime::Builder::new_current_thread().enable_all().build() {
            Ok(runtime) => runtime.block_on(async move {
                Self::release_entries(entries, context).await;
            }),
            Err(err) => warn!(context, "failed to create fallback unlock runtime: {}", err),
        });
        drop(join_handle);
    }

    fn spawn_pending_cleanup(
        mut pending: JoinSet<LockAcquireTaskResult>,
        clients: Vec<Arc<dyn LockClient>>,
        fallback_lock_id: LockId,
        context: &'static str,
    ) {
        let handle = tokio::spawn(async move {
            while let Some(join_result) = pending.join_next().await {
                match join_result {
                    Ok((idx, Ok(resp))) if resp.success => {
                        let lock_id = resp
                            .lock_info
                            .as_ref()
                            .map(|info| info.id.clone())
                            .unwrap_or_else(|| fallback_lock_id.clone());
                        let Some(client) = clients.get(idx) else {
                            tracing::warn!("{context}: missing client for pending lock cleanup at index {}", idx);
                            continue;
                        };

                        Self::release_entries(vec![(lock_id, client.clone())], context).await;
                    }
                    Ok((idx, Ok(resp))) => {
                        tracing::debug!(
                            "{context}: pending lock request on client {} completed without success: {:?}",
                            idx,
                            resp.error
                        );
                    }
                    Ok((idx, Err(err))) => {
                        tracing::warn!("{context}: pending lock request on client {} failed: {}", idx, err);
                    }
                    Err(err) => {
                        tracing::warn!("{context}: pending lock cleanup task join failed: {}", err);
                    }
                }
            }
        });
        drop(handle);
    }

    fn log_failed_lock_response(&self, request: &LockRequest, idx: usize, error: String) {
        if request.suppress_contention_logs || !should_warn_lock_failure(&error) {
            tracing::debug!(
                resource = %request.resource,
                owner = %request.owner,
                "Failed to acquire lock on client from response: {}, error: {}",
                idx,
                error
            );
        } else {
            tracing::warn!(
                resource = %request.resource,
                owner = %request.owner,
                "Failed to acquire lock on client from response: {}, error: {}",
                idx,
                error
            );
        }
    }

    async fn acquire_lock_quorum_with_retry(&self, request: &LockRequest) -> Result<LockAcquireQuorumResult> {
        let start = std::time::Instant::now();
        let mut attempt = 1;
        let mut last_result = None;

        loop {
            let elapsed = start.elapsed();
            if elapsed >= request.acquire_timeout {
                break;
            }

            let remaining = request.acquire_timeout - elapsed;
            let mut attempt_request = request.clone();
            attempt_request.acquire_timeout = self.lock_acquire_attempt_timeout(remaining);
            attempt_request.lock_id = LockId::new_unique(&request.resource);

            let result = self.acquire_lock_quorum_once(&attempt_request).await?;
            if result.response.success || !matches!(result.failure_kind, Some(LockAcquireFailureKind::RetryableContention)) {
                return Ok(result);
            }

            last_result = Some(result);
            let backoff = Self::lock_acquire_retry_backoff(attempt, &mut rand::rng());
            if start.elapsed().saturating_add(backoff) >= request.acquire_timeout {
                break;
            }

            tokio::time::sleep(backoff).await;
            attempt += 1;
        }

        if let Some(mut result) = last_result {
            if result.quorum_impossible {
                result.failure_kind = Some(LockAcquireFailureKind::UnrecoverableQuorum);
            }
            Ok(result)
        } else {
            Ok(LockAcquireQuorumResult {
                response: LockResponse::failure("Lock acquisition timeout", request.acquire_timeout),
                individual_locks: Vec::new(),
                failure_kind: Some(LockAcquireFailureKind::RetryableContention),
                quorum_impossible: false,
            })
        }
    }

    fn lock_acquire_timeout_result(timeout: Duration) -> LockAcquireQuorumResult {
        LockAcquireQuorumResult {
            response: LockResponse::failure("Lock acquisition timeout", timeout),
            individual_locks: Vec::new(),
            failure_kind: Some(LockAcquireFailureKind::RetryableContention),
            quorum_impossible: false,
        }
    }

    fn lock_acquire_attempt_timeout_result(
        timeout: Duration,
        individual_locks: Vec<HeldLockEntry>,
        last_failure: Option<String>,
        last_failure_kind: Option<LockAcquireFailureKind>,
    ) -> LockAcquireQuorumResult {
        let failure_kind =
            if let Some(kind @ (LockAcquireFailureKind::NonRetryable | LockAcquireFailureKind::UnrecoverableQuorum)) =
                last_failure_kind
            {
                kind
            } else {
                return Self::lock_acquire_timeout_result(timeout);
            };

        let mut error = "Lock acquisition timeout".to_string();
        if let Some(last_failure) = last_failure {
            error.push_str("; last failure: ");
            error.push_str(&last_failure);
        }

        LockAcquireQuorumResult {
            response: LockResponse::failure(error, timeout),
            individual_locks,
            failure_kind: Some(failure_kind),
            quorum_impossible: matches!(failure_kind, LockAcquireFailureKind::UnrecoverableQuorum),
        }
    }

    /// Quorum-based lock acquisition: success if at least the required quorum succeeds.
    /// Collects all individual lock_ids from successful clients and creates an aggregate lock_id.
    /// Returns the LockResponse with aggregate lock_id and individual lock mappings.
    async fn acquire_lock_quorum_once(&self, request: &LockRequest) -> Result<LockAcquireQuorumResult> {
        let required_quorum = self.required_quorum(request.lock_type);
        let attempt_started = Instant::now();
        let mut pending = self.spawn_lock_requests(request);
        let mut individual_locks: Vec<HeldLockEntry> = Vec::new();
        let fallback_lock_id = request.lock_id.clone();
        let mut last_failure = None;
        let mut last_failure_kind = None;
        let mut last_hard_failure_kind = None;
        let mut hard_failures = 0usize;
        let start = Instant::now();

        while !pending.is_empty() {
            let remaining = request.acquire_timeout.saturating_sub(start.elapsed());
            if remaining.is_zero() {
                Self::spawn_release_cleanup(
                    individual_locks.iter().map(HeldLockEntry::release_entry).collect(),
                    "distributed_lock_attempt_timeout",
                );
                Self::spawn_pending_cleanup(
                    pending,
                    self.clients.clone(),
                    fallback_lock_id.clone(),
                    "distributed_lock_attempt_timeout_cleanup",
                );
                return Ok(Self::lock_acquire_attempt_timeout_result(
                    request.acquire_timeout,
                    individual_locks,
                    last_failure,
                    last_failure_kind,
                ));
            }

            let join_result = match tokio::time::timeout(remaining, pending.join_next()).await {
                Ok(Some(join_result)) => join_result,
                Ok(None) => break,
                Err(_) => {
                    Self::spawn_release_cleanup(
                        individual_locks.iter().map(HeldLockEntry::release_entry).collect(),
                        "distributed_lock_attempt_timeout",
                    );
                    Self::spawn_pending_cleanup(
                        pending,
                        self.clients.clone(),
                        fallback_lock_id.clone(),
                        "distributed_lock_attempt_timeout_cleanup",
                    );
                    return Ok(Self::lock_acquire_attempt_timeout_result(
                        request.acquire_timeout,
                        individual_locks,
                        last_failure,
                        last_failure_kind,
                    ));
                }
            };

            match join_result {
                Ok((idx, Ok(resp))) => {
                    if resp.success {
                        let lock_id = resp
                            .lock_info
                            .as_ref()
                            .map(|info| info.id.clone())
                            .unwrap_or_else(|| fallback_lock_id.clone());

                        if let Some(client) = self.clients.get(idx) {
                            individual_locks.push(HeldLockEntry {
                                lock_id,
                                client: client.clone(),
                                valid_until: attempt_started.checked_add(request.ttl).unwrap_or(attempt_started),
                            });
                        } else {
                            tracing::warn!("Missing lock client at index {} while recording success", idx);
                        }
                    } else {
                        let error = resp.error.unwrap_or_else(|| "unknown error".to_string());
                        let failure_kind = classify_lock_failure(&error);
                        if is_remote_lock_rpc_failure(&error) && !is_remote_lock_rpc_timeout(&error) {
                            hard_failures += 1;
                            last_hard_failure_kind = Some(failure_kind);
                        }
                        self.log_failed_lock_response(request, idx, error.clone());
                        last_failure = Some(error);
                        last_failure_kind = Some(failure_kind);
                    }
                }
                Ok((idx, Err(err))) => {
                    hard_failures += 1;
                    let failure_kind = classify_lock_error_failure(&err);
                    tracing::warn!("Failed to acquire lock on client {}: {}", idx, err);
                    last_failure = Some(err.to_string());
                    last_failure_kind = Some(failure_kind);
                    last_hard_failure_kind = Some(failure_kind);
                }
                Err(err) => {
                    hard_failures += 1;
                    tracing::warn!("Lock acquisition task join failed: {}", err);
                    last_failure = Some(err.to_string());
                    last_failure_kind = Some(LockAcquireFailureKind::NonRetryable);
                    last_hard_failure_kind = Some(LockAcquireFailureKind::NonRetryable);
                }
            }

            if self.clients.len().saturating_sub(hard_failures) < required_quorum {
                let rollback_count = individual_locks.len();
                Self::spawn_release_cleanup(
                    individual_locks.iter().map(HeldLockEntry::release_entry).collect(),
                    "distributed_lock_quorum_rollback",
                );
                if !pending.is_empty() {
                    Self::spawn_pending_cleanup(
                        pending,
                        self.clients.clone(),
                        fallback_lock_id.clone(),
                        "distributed_lock_failure_cleanup",
                    );
                }

                let mut error = format!(
                    "Unrecoverable quorum failure: {rollback_count}/{required_quorum} required; {hard_failures} clients failed"
                );
                if let Some(last_failure) = last_failure {
                    error.push_str("; last failure: ");
                    error.push_str(&last_failure);
                }
                let resp = LockResponse::failure(error, Duration::ZERO);
                let failure_kind = last_hard_failure_kind.unwrap_or(LockAcquireFailureKind::UnrecoverableQuorum);
                return Ok(LockAcquireQuorumResult {
                    response: resp,
                    individual_locks,
                    failure_kind: Some(failure_kind),
                    quorum_impossible: true,
                });
            }

            if individual_locks.len() >= required_quorum {
                if !pending.is_empty() {
                    Self::spawn_pending_cleanup(
                        pending,
                        self.clients.clone(),
                        fallback_lock_id.clone(),
                        "distributed_lock_success_cleanup",
                    );
                }

                let aggregate_lock_id = generate_aggregate_lock_id(&request.resource);
                tracing::debug!(
                    "Generated aggregate lock_id {} for {} individual locks on resource {}",
                    aggregate_lock_id,
                    individual_locks.len(),
                    request.resource
                );

                let resp = LockResponse::success(
                    LockInfo {
                        id: aggregate_lock_id,
                        resource: request.resource.clone(),
                        lock_type: request.lock_type,
                        status: LockStatus::Acquired,
                        owner: request.owner.clone(),
                        acquired_at: std::time::SystemTime::now(),
                        expires_at: std::time::SystemTime::now() + request.ttl,
                        last_refreshed: std::time::SystemTime::now(),
                        metadata: request.metadata.clone(),
                        priority: request.priority,
                        wait_start_time: None,
                    },
                    Duration::ZERO,
                );
                return Ok(LockAcquireQuorumResult {
                    response: resp,
                    individual_locks,
                    failure_kind: None,
                    quorum_impossible: false,
                });
            }

            if individual_locks.len() + pending.len() < required_quorum {
                let rollback_count = individual_locks.len();
                Self::spawn_release_cleanup(
                    individual_locks.iter().map(HeldLockEntry::release_entry).collect(),
                    "distributed_lock_quorum_rollback",
                );
                if !pending.is_empty() {
                    Self::spawn_pending_cleanup(
                        pending,
                        self.clients.clone(),
                        fallback_lock_id.clone(),
                        "distributed_lock_failure_cleanup",
                    );
                }

                let mut error = format!("Failed to acquire quorum: {rollback_count}/{required_quorum} required");
                let (failure_kind, quorum_impossible) = if hard_failures > 0 {
                    error = format!(
                        "Unrecoverable quorum failure: {rollback_count}/{required_quorum} required; {hard_failures} clients failed; {error}"
                    );
                    (last_hard_failure_kind.unwrap_or(LockAcquireFailureKind::UnrecoverableQuorum), true)
                } else {
                    (LockAcquireFailureKind::RetryableContention, false)
                };
                if let Some(last_failure) = last_failure {
                    error.push_str("; last failure: ");
                    error.push_str(&last_failure);
                }
                let resp = LockResponse::failure(error, Duration::ZERO);
                return Ok(LockAcquireQuorumResult {
                    response: resp,
                    individual_locks,
                    failure_kind: Some(failure_kind),
                    quorum_impossible,
                });
            }
        }

        let rollback_count = individual_locks.len();
        Self::spawn_release_cleanup(
            individual_locks.iter().map(HeldLockEntry::release_entry).collect(),
            "distributed_lock_quorum_rollback",
        );
        let mut error = format!("Failed to acquire quorum: {rollback_count}/{required_quorum} required");
        if let Some(last_failure) = &last_failure {
            error.push_str("; last failure: ");
            error.push_str(last_failure);
        }
        let resp = LockResponse::failure(error, Duration::ZERO);
        Ok(LockAcquireQuorumResult {
            response: resp,
            individual_locks,
            failure_kind: Some(if hard_failures > 0 {
                last_hard_failure_kind.unwrap_or(LockAcquireFailureKind::UnrecoverableQuorum)
            } else {
                last_failure_kind.unwrap_or(LockAcquireFailureKind::RetryableContention)
            }),
            quorum_impossible: hard_failures > 0,
        })
    }
}

#[inline(always)]
fn record_lock_held_acquire(lock_type: LockType) {
    match lock_type {
        LockType::Shared => record_read_lock_held_acquire(),
        LockType::Exclusive => record_write_lock_held_acquire(),
    }
}

#[inline(always)]
fn record_lock_held_release(lock_type: LockType) {
    match lock_type {
        LockType::Shared => record_read_lock_held_release(),
        LockType::Exclusive => record_write_lock_held_release(),
    }
}

#[cfg(test)]
mod tests {
    use super::{
        DistributedLock, LOCK_ACQUIRE_ATTEMPT_TIMEOUT, LOCK_ACQUIRE_RETRY_INITIAL_BACKOFF, LockAcquireFailureKind,
        LockLostSignal, is_remote_lock_rpc_failure, should_warn_lock_failure,
    };
    use crate::{LockError, LockId, LockInfo, LockRequest, LockResponse, LockStats, LockType, ObjectKey, client::LockClient};
    use rand::{SeedableRng as _, TryRng, rngs::StdRng};
    use std::assert_matches;
    use std::convert::Infallible;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::{
        collections::{HashMap, VecDeque},
        sync::{Arc, Mutex},
        time::Duration,
    };

    #[derive(Clone, Copy, Debug)]
    enum RefreshOutcome {
        Alive,               // Ok(true)  refresh succeeded
        SlowAlive(Duration), // Ok(true)  after a delayed response
        NotFound,            // Ok(false) remote already lost the lock (reclaimed / never held)
        RpcError,            // Err        RPC jitter
    }

    struct ConstantRng(u64);

    impl TryRng for ConstantRng {
        type Error = Infallible;

        fn try_next_u32(&mut self) -> Result<u32, Self::Error> {
            Ok(self.0 as u32)
        }

        fn try_next_u64(&mut self) -> Result<u64, Self::Error> {
            Ok(self.0)
        }

        fn try_fill_bytes(&mut self, dst: &mut [u8]) -> Result<(), Self::Error> {
            for chunk in dst.chunks_mut(std::mem::size_of::<u64>()) {
                chunk.copy_from_slice(&self.0.to_ne_bytes()[..chunk.len()]);
            }
            Ok(())
        }
    }

    #[test]
    fn lock_acquire_retry_backoff_uses_bounded_jitter() {
        let mut rng = StdRng::seed_from_u64(42);
        let base = LOCK_ACQUIRE_RETRY_INITIAL_BACKOFF * 3;
        let jitter = base / 4;
        let lower = base - jitter;
        let upper = base + jitter;
        let samples = (0..32)
            .map(|_| DistributedLock::lock_acquire_retry_backoff(3, &mut rng))
            .collect::<Vec<_>>();

        assert!(samples.iter().all(|delay| *delay >= lower && *delay <= upper));
        assert!(samples.windows(2).any(|pair| pair[0] != pair[1]));

        let mut lower_rng = ConstantRng(0);
        let mut upper_rng = ConstantRng(u64::MAX);
        assert_eq!(DistributedLock::lock_acquire_retry_backoff(3, &mut lower_rng), lower);
        assert_eq!(DistributedLock::lock_acquire_retry_backoff(3, &mut upper_rng), upper);
    }

    /// Counting test client: acquires successfully and echoes back request.lock_id as
    /// the per-client id, so the guard heartbeat calls refresh with that id; refresh
    /// increments a counter and returns per the configured outcome.
    #[derive(Debug)]
    struct RefreshCountingClient {
        refresh_calls: Arc<AtomicUsize>,
        outcome: RefreshOutcome,
        active: tokio::sync::Mutex<std::collections::HashSet<LockId>>,
    }

    impl RefreshCountingClient {
        fn new(outcome: RefreshOutcome, counter: Arc<AtomicUsize>) -> Arc<Self> {
            Arc::new(Self {
                refresh_calls: counter,
                outcome,
                active: tokio::sync::Mutex::new(std::collections::HashSet::new()),
            })
        }
    }

    #[async_trait::async_trait]
    impl LockClient for RefreshCountingClient {
        async fn acquire_lock(&self, request: &LockRequest) -> crate::Result<LockResponse> {
            self.active.lock().await.insert(request.lock_id.clone());
            let info = LockInfo {
                id: request.lock_id.clone(),
                resource: request.resource.clone(),
                lock_type: request.lock_type,
                status: crate::types::LockStatus::Acquired,
                owner: request.owner.clone(),
                acquired_at: std::time::SystemTime::now(),
                expires_at: std::time::SystemTime::now() + request.ttl,
                last_refreshed: std::time::SystemTime::now(),
                metadata: request.metadata.clone(),
                priority: request.priority,
                wait_start_time: None,
            };
            Ok(LockResponse::success(info, Duration::ZERO))
        }
        async fn release(&self, lock_id: &LockId) -> crate::Result<bool> {
            Ok(self.active.lock().await.remove(lock_id))
        }
        async fn refresh(&self, _lock_id: &LockId) -> crate::Result<bool> {
            self.refresh_calls.fetch_add(1, Ordering::SeqCst);
            match self.outcome {
                RefreshOutcome::Alive => Ok(true),
                RefreshOutcome::SlowAlive(delay) => {
                    tokio::time::sleep(delay).await;
                    Ok(true)
                }
                RefreshOutcome::NotFound => Ok(false),
                RefreshOutcome::RpcError => Err(LockError::internal("refresh rpc jitter")),
            }
        }
        async fn force_release(&self, lock_id: &LockId) -> crate::Result<bool> {
            self.release(lock_id).await
        }
        async fn check_status(&self, _lock_id: &LockId) -> crate::Result<Option<LockInfo>> {
            Ok(None)
        }
        async fn get_stats(&self) -> crate::Result<LockStats> {
            Ok(LockStats::default())
        }
        async fn close(&self) -> crate::Result<()> {
            Ok(())
        }
        async fn is_online(&self) -> bool {
            true
        }
        async fn is_local(&self) -> bool {
            false
        }
    }

    fn counting_clients(outcomes: &[RefreshOutcome]) -> (Vec<Arc<dyn LockClient>>, Vec<Arc<AtomicUsize>>) {
        let mut clients: Vec<Arc<dyn LockClient>> = Vec::new();
        let mut counters = Vec::new();
        for outcome in outcomes {
            let counter = Arc::new(AtomicUsize::new(0));
            clients.push(RefreshCountingClient::new(*outcome, counter.clone()) as Arc<dyn LockClient>);
            counters.push(counter);
        }
        (clients, counters)
    }

    // A1 -- heartbeat broadcasts refresh every interval; a live lock stays refreshed (#899 keepalive).
    #[tokio::test]
    async fn heartbeat_keeps_distributed_lock_refreshed() {
        let (clients, counters) = counting_clients(&[
            RefreshOutcome::Alive,
            RefreshOutcome::Alive,
            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_acquire_timeout(Duration::from_millis(200))
            .with_ttl(Duration::from_millis(120))
            .with_refresh_interval(Duration::from_millis(20)); // interval < ttl -> spawn

        let guard = lock
            .acquire_guard(&request)
            .await
            .expect("acquire should not error")
            .expect("quorum should be reached");

        tokio::time::sleep(Duration::from_millis(90)).await; // spans >2 intervals

        // The guard tracks exactly the quorum of per-client locks (excess successes are cleaned
        // up on acquire), so the heartbeat broadcasts to at least `quorum` clients.
        let refreshed_clients = counters.iter().filter(|c| c.load(Ordering::SeqCst) >= 1).count();
        assert!(
            refreshed_clients >= 3,
            "heartbeat should refresh at least the quorum of tracked clients, got {refreshed_clients}"
        );
        assert!(!guard.is_lock_lost(), "all-alive refresh must not signal lock loss");
        drop(guard);
    }

    // A2 -- a quorum-one distributed lease still requires renewal.
    #[tokio::test]
    async fn single_client_guard_keeps_backend_lease_refreshed() {
        let (clients, counters) = counting_clients(&[RefreshOutcome::Alive]);
        let lock = DistributedLock::new("test".to_string(), clients, 1);
        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
            .expect("single-client lock acquisition should not error")
            .expect("single-client lock quorum should be reached");
        assert!(
            guard.refresh_task.is_some(),
            "single-client distributed guard must spawn a background task"
        );
        tokio::time::timeout(Duration::from_secs(2), async {
            while counters[0].load(Ordering::SeqCst) < 2 {
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        })
        .await
        .expect("single-client distributed guard should renew before the test deadline");

        assert!(
            counters[0].load(Ordering::SeqCst) >= 2,
            "single-client distributed guard must renew beyond the original backend TTL"
        );
        assert!(!guard.is_lock_lost(), "successful quorum-one refresh must preserve the lease");
        drop(guard);
    }

    // A3 -- dropping the guard aborts the heartbeat before release; no refresh lands after drop.
    #[tokio::test]
    async fn dropping_guard_stops_heartbeat() {
        let (clients, counters) = counting_clients(&[
            RefreshOutcome::Alive,
            RefreshOutcome::Alive,
            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
            .expect("distributed lock acquisition should not error")
            .expect("distributed lock quorum should be reached");
        tokio::time::sleep(Duration::from_millis(50)).await;
        drop(guard); // should abort the heartbeat first
        let after_drop: Vec<usize> = counters.iter().map(|c| c.load(Ordering::SeqCst)).collect();

        tokio::time::sleep(Duration::from_millis(80)).await; // wait >2 more intervals
        for (idx, counter) in counters.iter().enumerate() {
            assert_eq!(
                counter.load(Ordering::SeqCst),
                after_drop[idx],
                "client {idx} must not be refreshed after guard drop"
            );
        }
    }

    // A4 -- losing refresh quorum (not_found > n - refresh_quorum) signals lock loss.
    // 4 clients / write-quorum 3 -> n-quorum = 1; 2 NotFound (>1) -> lost.
    #[tokio::test]
    async fn heartbeat_quorum_loss_signals_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
            .expect("distributed lock acquisition should not error")
            .expect("distributed lock quorum should be reached");
        tokio::time::sleep(Duration::from_millis(60)).await; // at least one tick

        assert!(
            guard.is_lock_lost(),
            "losing refresh quorum (2 not_found of 4, quorum 3) must signal lock loss"
        );
        let signal = guard.lock_lost();
        tokio::time::timeout(Duration::from_millis(20), signal.notified())
            .await
            .expect("lock_lost().notified() must resolve once quorum lost");
        drop(guard);
    }

    #[tokio::test]
    async fn lock_lost_notification_cannot_be_missed_after_waiter_registration() {
        let signal = LockLostSignal::default();

        tokio::time::timeout(Duration::from_secs(2), signal.notified_after_registration(|| signal.mark_lost()))
            .await
            .expect("loss between waiter registration and state recheck must still resolve");
    }

    #[tokio::test]
    async fn expired_lease_is_fenced_without_waiting_for_monitor_scheduling() {
        let signal = LockLostSignal::default();
        signal.set_valid_until(tokio::time::Instant::now());

        assert!(signal.is_lost(), "an elapsed backend deadline must fence the owner immediately");
        tokio::time::timeout(Duration::from_secs(2), signal.notified())
            .await
            .expect("an elapsed backend deadline must resolve lock-loss waiters immediately");
    }

    #[tokio::test]
    async fn deadline_notification_permanently_fences_the_signal() {
        let signal = LockLostSignal::default();
        signal.set_valid_until(tokio::time::Instant::now() + Duration::from_millis(10));

        tokio::time::timeout(Duration::from_secs(2), signal.notified())
            .await
            .expect("the backend deadline must resolve lock-loss waiters");

        assert!(
            signal.lost.load(Ordering::SeqCst),
            "deadline notification must persist the lock-loss fence"
        );
    }

    #[test]
    fn expired_lease_cannot_be_revived_by_a_later_deadline() {
        let signal = LockLostSignal::default();
        signal.set_valid_until(tokio::time::Instant::now());
        assert!(signal.is_lost(), "an elapsed lease must become permanently fenced");

        signal.set_valid_until(tokio::time::Instant::now() + Duration::from_secs(60));

        assert!(signal.is_lost(), "a later heartbeat must not revive an expired lease");
    }

    // 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
            .expect("namespace lock acquisition should not error")
            .expect("namespace lock quorum should be reached");
        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 is tolerated within the lease, but an unconfirmed lease cannot
    // remain valid after its backend TTL.
    #[tokio::test]
    async fn heartbeat_rpc_error_expires_at_backend_ttl() {
        let (clients, _counters) = counting_clients(&[
            RefreshOutcome::RpcError,
            RefreshOutcome::RpcError,
            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
            .expect("distributed lock acquisition should not error")
            .expect("distributed lock quorum should be reached");
        tokio::time::sleep(Duration::from_millis(60)).await;

        assert!(
            !guard.is_lock_lost(),
            "transient RPC jitter before the backend TTL must not immediately signal lock loss"
        );

        let signal = guard.lock_lost();
        tokio::time::timeout(Duration::from_secs(2), signal.notified())
            .await
            .expect("unconfirmed leases must signal lock loss when their backend TTL expires");
        assert!(guard.is_lock_lost(), "the guard must fail closed once refresh quorum expires");
        drop(guard);
    }

    // A6 -- a refresh call that remains in flight cannot keep an expired quorum alive.
    #[tokio::test]
    async fn heartbeat_slow_refresh_is_fenced_by_backend_ttl() {
        let (clients, _counters) =
            counting_clients(&[RefreshOutcome::SlowAlive(Duration::from_millis(200)), RefreshOutcome::Alive]);
        let lock = DistributedLock::new("test".to_string(), clients, 2);
        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
            .expect("distributed lock acquisition should not error")
            .expect("distributed lock quorum should be reached");
        let signal = guard.lock_lost();
        tokio::time::timeout(Duration::from_secs(2), signal.notified())
            .await
            .expect("an in-flight refresh must not extend an unconfirmed lease past its TTL");
        assert!(
            guard.is_lock_lost(),
            "slow refresh must fail closed at the last confirmed quorum deadline"
        );
        drop(guard);
    }

    // A7 -- sub-second ttl boundary: no panic, and skip heartbeat when interval>=ttl
    //        (fixes clamp panic; aligns with the 50ms-ttl distributed path in namespace tests).
    #[tokio::test]
    async fn subsecond_ttl_guard_does_not_panic_and_skips_heartbeat() {
        // tiny ttl; the point is that acquire must not panic.
        let (clients, _counters) = counting_clients(&[RefreshOutcome::Alive, RefreshOutcome::Alive, RefreshOutcome::Alive]);
        let lock = DistributedLock::new("test".to_string(), clients, 2);
        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
            .with_acquire_timeout(Duration::from_millis(100))
            .with_ttl(Duration::from_millis(50)); // a clamp(1s, ttl-1s) would panic here
        let guard = lock
            .acquire_guard(&request)
            .await
            .expect("subsecond ttl acquire must not panic")
            .expect("quorum should be reached");
        drop(guard);

        // inject interval >= ttl -> degenerate to no spawn
        let (clients2, counters2) = counting_clients(&[RefreshOutcome::Alive, RefreshOutcome::Alive, RefreshOutcome::Alive]);
        let lock2 = DistributedLock::new("test".to_string(), clients2, 2);
        let request2 = LockRequest::new(ObjectKey::new("bucket", "object2"), LockType::Exclusive, "owner")
            .with_ttl(Duration::from_millis(50))
            .with_refresh_interval(Duration::from_millis(50)); // interval >= ttl -> None
        let guard2 = lock2
            .acquire_guard(&request2)
            .await
            .expect("subsecond lock acquisition should not error")
            .expect("subsecond lock quorum should be reached");
        assert!(
            guard2.refresh_task.is_none(),
            "non-renewable subsecond guard must not spawn a background task"
        );
        tokio::time::timeout(Duration::from_secs(2), guard2.lock_lost().notified())
            .await
            .expect("non-renewable subsecond guard must signal loss at its backend TTL");
        for c in &counters2 {
            assert_eq!(c.load(Ordering::SeqCst), 0, "interval>=ttl must not refresh the backend lease");
        }
        assert!(guard2.is_lock_lost(), "interval>=ttl guard must fail closed after backend TTL");
        drop(guard2);
    }

    // A8 -- disarm() must also stop the heartbeat (public API gap).
    #[tokio::test]
    async fn disarm_stops_heartbeat() {
        let (clients, counters) = counting_clients(&[
            RefreshOutcome::Alive,
            RefreshOutcome::Alive,
            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 mut guard = lock
            .acquire_guard(&request)
            .await
            .expect("distributed lock acquisition should not error")
            .expect("distributed lock quorum should be reached");
        tokio::time::sleep(Duration::from_millis(50)).await;
        guard.disarm(); // should abort the heartbeat
        let after_disarm: Vec<usize> = counters.iter().map(|c| c.load(Ordering::SeqCst)).collect();

        tokio::time::sleep(Duration::from_millis(80)).await;
        for (idx, counter) in counters.iter().enumerate() {
            assert_eq!(
                counter.load(Ordering::SeqCst),
                after_disarm[idx],
                "disarm() must stop the heartbeat for client {idx}"
            );
        }
        drop(guard);
    }

    #[derive(Debug)]
    struct ResponseClient {
        response: LockResponse,
        delay: Duration,
    }

    impl ResponseClient {
        fn new(response: LockResponse) -> Self {
            Self {
                response,
                delay: Duration::ZERO,
            }
        }

        fn with_delay(mut self, delay: Duration) -> Self {
            self.delay = delay;
            self
        }

        fn into_client(self) -> Arc<dyn LockClient> {
            Arc::new(self)
        }
    }

    #[async_trait::async_trait]
    impl LockClient for ResponseClient {
        async fn acquire_lock(&self, _request: &LockRequest) -> crate::Result<LockResponse> {
            if !self.delay.is_zero() {
                tokio::time::sleep(self.delay).await;
            }
            Ok(self.response.clone())
        }

        async fn release(&self, _lock_id: &LockId) -> crate::Result<bool> {
            Ok(false)
        }

        async fn refresh(&self, _lock_id: &LockId) -> crate::Result<bool> {
            Ok(false)
        }

        async fn force_release(&self, _lock_id: &LockId) -> crate::Result<bool> {
            Ok(false)
        }

        async fn check_status(&self, _lock_id: &LockId) -> crate::Result<Option<LockInfo>> {
            Ok(None)
        }

        async fn get_stats(&self) -> crate::Result<LockStats> {
            Ok(LockStats::default())
        }

        async fn close(&self) -> crate::Result<()> {
            Ok(())
        }

        async fn is_online(&self) -> bool {
            true
        }

        async fn is_local(&self) -> bool {
            false
        }
    }

    #[derive(Debug)]
    struct TimeoutRecordingClient {
        seen_timeouts: Arc<Mutex<Vec<Duration>>>,
    }

    impl TimeoutRecordingClient {
        fn new(seen_timeouts: Arc<Mutex<Vec<Duration>>>) -> Self {
            Self { seen_timeouts }
        }

        fn into_client(self) -> Arc<dyn LockClient> {
            Arc::new(self)
        }
    }

    #[async_trait::async_trait]
    impl LockClient for TimeoutRecordingClient {
        async fn acquire_lock(&self, request: &LockRequest) -> crate::Result<LockResponse> {
            self.seen_timeouts.lock().unwrap().push(request.acquire_timeout);
            Ok(LockResponse::failure("Lock acquisition timeout", Duration::ZERO))
        }

        async fn release(&self, _lock_id: &LockId) -> crate::Result<bool> {
            Ok(false)
        }

        async fn refresh(&self, _lock_id: &LockId) -> crate::Result<bool> {
            Ok(false)
        }

        async fn force_release(&self, _lock_id: &LockId) -> crate::Result<bool> {
            Ok(false)
        }

        async fn check_status(&self, _lock_id: &LockId) -> crate::Result<Option<LockInfo>> {
            Ok(None)
        }

        async fn get_stats(&self) -> crate::Result<LockStats> {
            Ok(LockStats::default())
        }

        async fn close(&self) -> crate::Result<()> {
            Ok(())
        }

        async fn is_online(&self) -> bool {
            true
        }

        async fn is_local(&self) -> bool {
            false
        }
    }

    #[derive(Debug, Clone, Copy)]
    enum AcquirePlan {
        Success { delay: Duration },
        Failure { error: &'static str, delay: Duration },
        ClientError { message: &'static str, delay: Duration },
    }

    #[derive(Debug)]
    struct SequencedClient {
        plans: Mutex<VecDeque<AcquirePlan>>,
        active: tokio::sync::Mutex<HashMap<LockId, LockInfo>>,
        seen_ids: Arc<Mutex<Vec<LockId>>>,
    }

    impl SequencedClient {
        fn new(plans: Vec<AcquirePlan>, seen_ids: Arc<Mutex<Vec<LockId>>>) -> Self {
            Self {
                plans: Mutex::new(plans.into()),
                active: tokio::sync::Mutex::new(HashMap::new()),
                seen_ids,
            }
        }
    }

    #[async_trait::async_trait]
    impl LockClient for SequencedClient {
        async fn acquire_lock(&self, request: &LockRequest) -> crate::Result<LockResponse> {
            self.seen_ids.lock().unwrap().push(request.lock_id.clone());

            let plan = self
                .plans
                .lock()
                .unwrap()
                .pop_front()
                .unwrap_or(AcquirePlan::Success { delay: Duration::ZERO });

            match plan {
                AcquirePlan::Success { delay } => {
                    if !delay.is_zero() {
                        tokio::time::sleep(delay).await;
                    }

                    let lock_info = LockInfo {
                        id: request.lock_id.clone(),
                        resource: request.resource.clone(),
                        lock_type: request.lock_type,
                        status: crate::types::LockStatus::Acquired,
                        owner: request.owner.clone(),
                        acquired_at: std::time::SystemTime::now(),
                        expires_at: std::time::SystemTime::now() + request.ttl,
                        last_refreshed: std::time::SystemTime::now(),
                        metadata: request.metadata.clone(),
                        priority: request.priority,
                        wait_start_time: None,
                    };

                    self.active.lock().await.insert(request.lock_id.clone(), lock_info.clone());

                    Ok(LockResponse::success(lock_info, Duration::ZERO))
                }
                AcquirePlan::Failure { error, delay } => {
                    if !delay.is_zero() {
                        tokio::time::sleep(delay).await;
                    }
                    Ok(LockResponse::failure(error, Duration::ZERO))
                }
                AcquirePlan::ClientError { message, delay } => {
                    if !delay.is_zero() {
                        tokio::time::sleep(delay).await;
                    }
                    Err(LockError::internal(message))
                }
            }
        }

        async fn release(&self, lock_id: &LockId) -> crate::Result<bool> {
            Ok(self.active.lock().await.remove(lock_id).is_some())
        }

        async fn refresh(&self, _lock_id: &LockId) -> crate::Result<bool> {
            Ok(false)
        }

        async fn force_release(&self, lock_id: &LockId) -> crate::Result<bool> {
            self.release(lock_id).await
        }

        async fn check_status(&self, lock_id: &LockId) -> crate::Result<Option<LockInfo>> {
            Ok(self.active.lock().await.get(lock_id).cloned())
        }

        async fn get_stats(&self) -> crate::Result<LockStats> {
            Ok(LockStats::default())
        }

        async fn close(&self) -> crate::Result<()> {
            Ok(())
        }

        async fn is_online(&self) -> bool {
            true
        }

        async fn is_local(&self) -> bool {
            false
        }
    }

    #[test]
    fn test_is_remote_lock_rpc_failure() {
        assert!(is_remote_lock_rpc_failure("Remote lock RPC failed: backend unreachable"));
        assert!(is_remote_lock_rpc_failure("remote lock rpc failed: temporary network issue"));
        assert!(is_remote_lock_rpc_failure("Remote lock RPC timed out: RPC timed out after 50ms"));
        assert!(!is_remote_lock_rpc_failure("Lock is already held"));
        assert!(!is_remote_lock_rpc_failure("acquired lock failed for other reason"));
    }

    #[test]
    fn test_should_warn_lock_failure() {
        assert!(should_warn_lock_failure("Remote lock RPC failed: backend unreachable"));
        assert!(should_warn_lock_failure("permission denied"));
        assert!(should_warn_lock_failure("Unrecoverable quorum failure: 1/3 required"));
        assert!(!should_warn_lock_failure("Remote lock RPC timed out: RPC timed out after 50ms"));
        assert!(!should_warn_lock_failure("Lock acquisition timeout"));
    }

    #[tokio::test]
    async fn acquire_guard_returns_quorum_error_when_rpc_failures_make_quorum_impossible() {
        let clients: Vec<Arc<dyn LockClient>> = vec![
            ResponseClient::new(LockResponse::failure("Remote lock RPC failed: node unavailable", Duration::ZERO)).into_client(),
            ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO))
                .with_delay(Duration::from_millis(50))
                .into_client(),
            ResponseClient::new(LockResponse::failure("Remote lock RPC failed: connection refused", Duration::ZERO))
                .into_client(),
            ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO))
                .with_delay(Duration::from_millis(50))
                .into_client(),
        ];
        let lock = DistributedLock::new("test".to_string(), clients, 3);
        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
            .with_acquire_timeout(Duration::from_secs(1));

        let result = lock.acquire_guard(&request).await;

        assert!(
            matches!(
                result,
                Err(LockError::QuorumNotReached {
                    required: 3,
                    achieved: 0
                })
            ),
            "unexpected result: {result:?}"
        );
    }

    #[tokio::test]
    async fn acquire_guard_retries_rpc_timeouts_until_caller_timeout() {
        let clients: Vec<Arc<dyn LockClient>> = vec![
            ResponseClient::new(LockResponse::failure(
                "Remote lock RPC timed out: RPC timed out after 50ms",
                Duration::ZERO,
            ))
            .into_client(),
            ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO))
                .with_delay(Duration::from_millis(50))
                .into_client(),
            ResponseClient::new(LockResponse::failure(
                "Remote lock RPC timed out: RPC timed out after 50ms",
                Duration::ZERO,
            ))
            .into_client(),
            ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO))
                .with_delay(Duration::from_millis(50))
                .into_client(),
        ];
        let lock = DistributedLock::new("test".to_string(), clients, 3);
        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
            .with_acquire_timeout(Duration::from_millis(900));

        let started = tokio::time::Instant::now();
        let result = lock.acquire_guard(&request).await;
        let elapsed = started.elapsed();

        assert_matches!(result, Ok(None));
        assert!(
            elapsed >= Duration::from_millis(250),
            "remote RPC timeouts should be retried within the caller timeout, got {elapsed:?}"
        );
    }

    #[tokio::test]
    async fn acquire_guard_retries_remote_rpc_failures_that_temporarily_preclude_quorum() {
        let clients: Vec<Arc<dyn LockClient>> = vec![
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::Failure {
                        error: "Remote lock RPC failed: node unavailable",
                        delay: Duration::ZERO,
                    },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                Arc::new(Mutex::new(Vec::new())),
            )),
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::Failure {
                        error: "Remote lock RPC failed: connection reset",
                        delay: Duration::ZERO,
                    },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                Arc::new(Mutex::new(Vec::new())),
            )),
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::Success { delay: Duration::ZERO },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                Arc::new(Mutex::new(Vec::new())),
            )),
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::Success { delay: Duration::ZERO },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                Arc::new(Mutex::new(Vec::new())),
            )),
        ];
        let lock = DistributedLock::new("test".to_string(), clients, 3);
        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
            .with_acquire_timeout(Duration::from_secs(2));

        let guard = lock
            .acquire_guard(&request)
            .await
            .expect("transient remote RPC quorum gap should retry")
            .expect("second attempt should acquire quorum");

        assert!(guard.entries.len() >= 3, "retry should acquire quorum entries");
        drop(guard);
    }

    #[tokio::test]
    async fn acquire_guard_retries_client_errors_that_temporarily_preclude_quorum() {
        let clients: Vec<Arc<dyn LockClient>> = vec![
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::ClientError {
                        message: "can not get client, err: temporary transport unavailable",
                        delay: Duration::ZERO,
                    },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                Arc::new(Mutex::new(Vec::new())),
            )),
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::ClientError {
                        message: "can not get client, err: connection pool busy",
                        delay: Duration::ZERO,
                    },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                Arc::new(Mutex::new(Vec::new())),
            )),
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::Success { delay: Duration::ZERO },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                Arc::new(Mutex::new(Vec::new())),
            )),
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::Success { delay: Duration::ZERO },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                Arc::new(Mutex::new(Vec::new())),
            )),
        ];
        let lock = DistributedLock::new("test".to_string(), clients, 3);
        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
            .with_acquire_timeout(Duration::from_secs(2));

        let guard = lock
            .acquire_guard(&request)
            .await
            .expect("retryable client error quorum gap should retry")
            .expect("second attempt should acquire quorum");

        assert!(guard.entries.len() >= 3, "retry should acquire quorum entries");
        drop(guard);
    }

    #[tokio::test]
    async fn acquire_guard_returns_timeout_when_zero_locks_make_quorum_impossible_for_attempt() {
        let clients: Vec<Arc<dyn LockClient>> = vec![
            ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)).into_client(),
            ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)).into_client(),
            ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO))
                .with_delay(Duration::from_secs(1))
                .into_client(),
            ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO))
                .with_delay(Duration::from_secs(1))
                .into_client(),
        ];
        let lock = DistributedLock::new("test".to_string(), clients, 3);
        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
            .with_acquire_timeout(Duration::from_millis(120));

        let started = tokio::time::Instant::now();
        let result = lock.acquire_guard(&request).await;

        assert_matches!(result, Ok(None));
        assert!(
            started.elapsed() < Duration::from_secs(1),
            "acquire should fail this attempt before waiting for delayed impossible-quorum tasks"
        );
    }

    #[tokio::test]
    async fn acquire_guard_retries_attempt_timeout_when_hard_failure_does_not_preclude_quorum() {
        let clients: Vec<Arc<dyn LockClient>> = vec![
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::Failure {
                        error: "Remote lock RPC failed: node unavailable",
                        delay: Duration::ZERO,
                    },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                Arc::new(Mutex::new(Vec::new())),
            )),
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::Success {
                        delay: Duration::from_millis(1200),
                    },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                Arc::new(Mutex::new(Vec::new())),
            )),
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::Success {
                        delay: Duration::from_millis(1200),
                    },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                Arc::new(Mutex::new(Vec::new())),
            )),
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::Success {
                        delay: Duration::from_millis(1200),
                    },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                Arc::new(Mutex::new(Vec::new())),
            )),
        ];
        let lock = DistributedLock::new("test".to_string(), clients, 3);
        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
            .with_acquire_timeout(Duration::from_secs(2));

        let guard = lock
            .acquire_guard(&request)
            .await
            .expect("attempt timeout with possible quorum should retry")
            .expect("second attempt should acquire quorum");

        assert_eq!(guard.entries.len(), 3);
        drop(guard);
    }

    #[tokio::test]
    async fn acquire_quorum_timeout_preserves_non_retryable_failure() {
        let clients: Vec<Arc<dyn LockClient>> = vec![
            ResponseClient::new(LockResponse::failure("permission denied", Duration::ZERO)).into_client(),
            ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO))
                .with_delay(Duration::from_secs(1))
                .into_client(),
            ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO))
                .with_delay(Duration::from_secs(1))
                .into_client(),
        ];
        let lock = DistributedLock::new("test".to_string(), clients, 2);
        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
            .with_acquire_timeout(Duration::from_millis(120));

        let result = lock.acquire_lock_quorum_with_retry(&request).await.unwrap();

        assert_matches!(result.failure_kind, Some(LockAcquireFailureKind::NonRetryable));
        assert!(
            result
                .response
                .error
                .as_deref()
                .is_some_and(|error| error.contains("permission denied")),
            "unexpected response: {:?}",
            result.response
        );
    }

    #[tokio::test]
    async fn acquire_guard_retries_transient_timeout_before_quorum() {
        let clients: Vec<Arc<dyn LockClient>> = vec![
            ResponseClient::new(LockResponse::failure(
                "Timeout { resource: \"bucket/object-flaky-acquire@latest\", timeout: 1s }",
                Duration::ZERO,
            ))
            .into_client(),
            ResponseClient::new(LockResponse::failure(
                "Timeout { resource: \"bucket/object-flaky-acquire@latest\", timeout: 1s }",
                Duration::ZERO,
            ))
            .into_client(),
            ResponseClient::new(LockResponse::failure(
                "Timeout { resource: \"bucket/object-flaky-acquire@latest\", timeout: 1s }",
                Duration::ZERO,
            ))
            .into_client(),
            ResponseClient::new(LockResponse::failure(
                "Timeout { resource: \"bucket/object-flaky-acquire@latest\", timeout: 1s }",
                Duration::ZERO,
            ))
            .into_client(),
        ];
        let lock = DistributedLock::new("test".to_string(), clients, 3);
        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
            .with_acquire_timeout(Duration::from_millis(900));

        let started = tokio::time::Instant::now();
        let result = lock.acquire_guard(&request).await;
        let elapsed = started.elapsed();

        assert_matches!(result, Ok(None));
        assert!(
            elapsed >= Duration::from_millis(250),
            "expected at least one retry attempt for transient timeout, got {elapsed:?}"
        );
    }

    #[tokio::test]
    async fn acquire_guard_uses_bounded_distributed_attempt_timeout() {
        let seen_timeouts = Arc::new(Mutex::new(Vec::new()));
        let clients: Vec<Arc<dyn LockClient>> = (0..4)
            .map(|_| TimeoutRecordingClient::new(seen_timeouts.clone()).into_client())
            .collect();
        let lock = DistributedLock::new("test".to_string(), clients, 3);
        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
            .with_acquire_timeout(Duration::from_millis(1500));

        let result = lock.acquire_guard(&request).await;

        assert_matches!(result, Ok(None));
        let seen_timeouts = seen_timeouts.lock().unwrap();
        assert!(!seen_timeouts.is_empty(), "expected lock clients to observe acquire timeouts");
        assert!(
            seen_timeouts.iter().all(|timeout| *timeout <= LOCK_ACQUIRE_ATTEMPT_TIMEOUT),
            "distributed attempts should be bounded by {LOCK_ACQUIRE_ATTEMPT_TIMEOUT:?}, saw {seen_timeouts:?}"
        );
    }

    #[tokio::test]
    async fn acquire_guard_enforces_attempt_timeout_when_clients_ignore_budget() {
        let clients: Vec<Arc<dyn LockClient>> = (0..4)
            .map(|_| {
                ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO))
                    .with_delay(Duration::from_secs(5))
                    .into_client()
            })
            .collect();
        let lock = DistributedLock::new("test".to_string(), clients, 3);
        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
            .with_acquire_timeout(Duration::from_millis(200));

        let result = tokio::time::timeout(Duration::from_millis(800), lock.acquire_guard(&request)).await;

        assert_matches!(result, Ok(Ok(None)));
    }

    #[tokio::test]
    async fn acquire_guard_retries_partial_quorum_after_rollback() {
        let seen_ids = Arc::new(Mutex::new(Vec::new()));
        let clients: Vec<Arc<dyn LockClient>> = vec![
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::Success { delay: Duration::ZERO },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                seen_ids.clone(),
            )),
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::Failure {
                        error: "Lock acquisition timeout",
                        delay: Duration::ZERO,
                    },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                seen_ids.clone(),
            )),
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::Failure {
                        error: "Lock acquisition timeout",
                        delay: Duration::ZERO,
                    },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                seen_ids.clone(),
            )),
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::Failure {
                        error: "Lock acquisition timeout",
                        delay: Duration::ZERO,
                    },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                seen_ids.clone(),
            )),
        ];
        let lock = DistributedLock::new("test".to_string(), clients, 3);
        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
            .with_acquire_timeout(Duration::from_secs(2));

        let guard = lock
            .acquire_guard(&request)
            .await
            .expect("partial quorum retry should not fail")
            .expect("second attempt should reach quorum");

        let unique_id_count = seen_ids
            .lock()
            .unwrap()
            .iter()
            .fold(Vec::<String>::new(), |mut uuids, lock_id| {
                if !uuids.iter().any(|uuid| uuid == &lock_id.uuid) {
                    uuids.push(lock_id.uuid.clone());
                }
                uuids
            })
            .len();

        assert!(unique_id_count >= 2, "partial quorum should retry with a fresh lock id");
        drop(guard);
    }

    #[tokio::test(start_paused = true)]
    async fn acquire_guard_returns_timeout_when_quorum_remains_contended() {
        let clients: Vec<Arc<dyn LockClient>> = vec![
            ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)).into_client(),
            ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)).into_client(),
            ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)).into_client(),
            ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)).into_client(),
        ];
        let lock = DistributedLock::new("test".to_string(), clients, 3);
        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
            .with_acquire_timeout(Duration::from_millis(120));

        let result = tokio::time::timeout(Duration::from_secs(1), lock.acquire_guard(&request)).await;

        assert_matches!(result, Ok(Ok(None)));
    }

    #[tokio::test]
    async fn acquire_guard_uses_fresh_lock_ids_across_retry_attempts() {
        let seen_ids = Arc::new(Mutex::new(Vec::new()));
        let delayed_success_a = Arc::new(SequencedClient::new(
            vec![
                AcquirePlan::Success {
                    delay: Duration::from_millis(400),
                },
                AcquirePlan::Success { delay: Duration::ZERO },
            ],
            seen_ids.clone(),
        ));
        let delayed_success_b = Arc::new(SequencedClient::new(
            vec![
                AcquirePlan::Success {
                    delay: Duration::from_millis(400),
                },
                AcquirePlan::Success { delay: Duration::ZERO },
            ],
            seen_ids.clone(),
        ));

        let clients: Vec<Arc<dyn LockClient>> = vec![
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::Failure {
                        error: "Lock acquisition timeout",
                        delay: Duration::ZERO,
                    },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                seen_ids.clone(),
            )),
            Arc::new(SequencedClient::new(
                vec![
                    AcquirePlan::Failure {
                        error: "Lock acquisition timeout",
                        delay: Duration::ZERO,
                    },
                    AcquirePlan::Success { delay: Duration::ZERO },
                ],
                seen_ids.clone(),
            )),
            delayed_success_a.clone(),
            delayed_success_b.clone(),
        ];

        let lock = DistributedLock::new("test".to_string(), clients, 3);
        let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner")
            .with_acquire_timeout(Duration::from_secs(2));

        let guard = lock
            .acquire_guard(&request)
            .await
            .expect("retry path should not fail")
            .expect("second attempt should reach quorum");

        tokio::time::sleep(Duration::from_millis(700)).await;

        let mut unique_ids = Vec::new();
        for lock_id in seen_ids.lock().unwrap().iter() {
            if unique_ids.iter().all(|id: &LockId| id.uuid != lock_id.uuid) {
                unique_ids.push(lock_id.clone());
            }
        }

        assert!(
            unique_ids.len() >= 2,
            "expected retry attempts to use distinct lock ids, saw {unique_ids:?}"
        );

        let retry_lock_ids = unique_ids.iter().skip(1).cloned().collect::<Vec<_>>();
        assert!(
            !retry_lock_ids.is_empty(),
            "expected at least one retry-attempt lock id, saw {unique_ids:?}"
        );

        let delayed_a_active = delayed_success_a.active.lock().await;
        let delayed_b_active = delayed_success_b.active.lock().await;
        let remaining_delayed_lock_ids = delayed_a_active
            .keys()
            .chain(delayed_b_active.keys())
            .cloned()
            .collect::<Vec<_>>();

        assert!(
            remaining_delayed_lock_ids.len() == 1,
            "exactly one delayed client lock should remain held by the retry guard after late cleanup"
        );
        assert!(
            retry_lock_ids
                .iter()
                .any(|retry_lock_id| retry_lock_id.uuid == remaining_delayed_lock_ids[0].uuid),
            "late cleanup must not leave only a first-attempt delayed lock active"
        );
        drop(delayed_b_active);
        drop(delayed_a_active);

        drop(guard);
    }
}