// 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::cluster::rpc::client::{
    AuthenticatedChannel, TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client,
};
use crate::cluster::rpc::set_tonic_rolling_mutation_body_digest;
use async_trait::async_trait;
use bytes::Bytes;
use rustfs_lock::{
    LockClient, LockError, LockInfo, LockRequest, LockResponse, LockStats, LockStatus, LockType, Result,
    types::{LockId, LockMetadata, LockPriority},
};
use rustfs_protos::proto_gen::node_service::{BatchGenerallyLockRequest, GenerallyLockRequest, PingRequest};
use rustfs_protos::{
    ConnectionEvictionLogLevel, evict_failed_connection_with_log_level, models::PingBodyBuilder,
    proto_gen::node_service::node_service_client::NodeServiceClient,
};
use std::time::Duration;
use tokio::time::timeout;
use tonic::Request;
use tonic::service::interceptor::InterceptedService;
use tracing::{debug, info, warn};

fn attach_lock_mutation_body_digest<T: rustfs_protos::CanonicalMutationBody>(request: &mut Request<T>) -> std::io::Result<()> {
    set_tonic_rolling_mutation_body_digest(request)
}

/// Remote lock client implementation
#[derive(Debug, Clone)]
pub struct RemoteClient {
    addr: String,
}

impl RemoteClient {
    pub fn new(endpoint: String) -> Self {
        Self { addr: endpoint }
    }

    pub fn from_url(url: url::Url) -> Self {
        Self { addr: url.to_string() }
    }

    fn build_ping_request() -> PingRequest {
        let mut fbb = flatbuffers::FlatBufferBuilder::new();
        let payload = fbb.create_vector(b"health-check");
        let mut builder = PingBodyBuilder::new(&mut fbb);
        builder.add_payload(payload);
        let root = builder.finish();
        fbb.finish(root, None);

        PingRequest {
            version: 1,
            body: Bytes::copy_from_slice(fbb.finished_data()),
        }
    }

    /// Create a minimal LockRequest for unlock operations using only lock_id
    fn create_unlock_request(lock_id: &LockId) -> LockRequest {
        LockRequest {
            lock_id: lock_id.clone(),
            resource: lock_id.resource.clone(),
            lock_type: LockType::Exclusive, // Type doesn't matter for unlock
            owner: String::new(),           // Owner not needed, server uses lock_id
            acquire_timeout: std::time::Duration::from_secs(30),
            ttl: std::time::Duration::from_secs(300),
            metadata: LockMetadata::default(),
            priority: LockPriority::Normal,
            deadlock_detection: false,
            suppress_contention_logs: false,
            refresh_interval: None,
        }
    }

    pub async fn get_client(&self) -> Result<NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>> {
        // P3-2 offline bypass (now covering the lock path too): fast-fail a peer already marked
        // offline instead of paying the connect timeout, so dsync reaches quorum sooner. Does not
        // change quorum; the self-healing re-probe keeps the peer recoverable.
        if let Some(reason) = crate::cluster::rpc::remote_disk::internode_offline_bypass_reason(&self.addr) {
            return Err(LockError::internal(reason));
        }
        node_service_time_out_client(&self.addr, TonicInterceptor::Signature(gen_tonic_signature_interceptor()))
            .await
            .map_err(|err| LockError::internal(format!("can not get client, err: {err}")))
    }

    fn is_scanner_leader_lock(resource_summary: &str) -> bool {
        resource_summary == ".rustfs.sys/leader.lock@latest"
    }

    /// Classify a `tonic::Status` as a transport-level failure of the cached channel.
    ///
    /// Only genuine connection problems (connect refused/reset, keepalive/deadline
    /// on the channel itself, cancelled in-flight streams) justify evicting and
    /// re-dialing the cached channel. A broken transport surfaces either as one of
    /// the codes below or carries the underlying hyper/h2 error as its `source`.
    ///
    /// Server-produced application statuses (`Unauthenticated`/`PermissionDenied`
    /// from the signature interceptor, `Internal`/`FailedPrecondition` when the
    /// peer's lock service is not ready yet, `InvalidArgument`, `ResourceExhausted`,
    /// `Unimplemented`, ...) are reconstructed from grpc-status trailers on the
    /// client and have no `source`. Evicting the channel for those cannot help —
    /// the channel is healthy — and only churns the connection while advancing the
    /// peer toward the offline threshold. See issue #4567.
    fn is_transport_failure(status: &tonic::Status) -> bool {
        use tonic::Code;
        std::error::Error::source(status).is_some()
            || matches!(
                status.code(),
                Code::Unavailable | Code::DeadlineExceeded | Code::Unknown | Code::Cancelled
            )
    }

    async fn evict_connection(&self, op: &'static str, reason: &str, resource_summary: &str) {
        let log_level = if Self::is_scanner_leader_lock(resource_summary) {
            debug!(
                addr = %self.addr,
                op,
                reason,
                resource_summary,
                "Evicting cached remote lock connection for scanner leader-lock RPC failure"
            );
            ConnectionEvictionLogLevel::Debug
        } else {
            warn!(
                addr = %self.addr,
                op,
                reason,
                resource_summary,
                "Evicting cached remote lock connection after RPC failure"
            );
            ConnectionEvictionLogLevel::Warn
        };
        evict_failed_connection_with_log_level(&self.addr, log_level).await;
    }

    fn summarize_resources(requests: &[LockRequest]) -> String {
        const LIMIT: usize = 3;
        let mut resources = requests
            .iter()
            .take(LIMIT)
            .map(|request| request.resource.to_string())
            .collect::<Vec<_>>();
        if requests.len() > LIMIT {
            resources.push(format!("... (+{} more)", requests.len() - LIMIT));
        }
        resources.join(", ")
    }

    fn rpc_timeout() -> Duration {
        Duration::from_millis(
            rustfs_utils::get_env_u64(
                rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS,
                rustfs_config::DEFAULT_OBJECT_LOCK_RPC_TIMEOUT_MS,
            )
            .max(1),
        )
    }

    async fn execute_rpc<T, F>(&self, op: &'static str, resource_summary: &str, future: F) -> std::result::Result<T, LockError>
    where
        F: std::future::Future<Output = std::result::Result<T, tonic::Status>>,
    {
        let lock_timeout = Self::rpc_timeout();
        match timeout(lock_timeout, future).await {
            Ok(Ok(response)) => Ok(response),
            Ok(Err(err)) => {
                let reason = err.to_string();
                // Only evict (and re-dial) the cached channel when the failure is a genuine
                // transport problem. A server-produced application status (auth denied, peer
                // lock service not ready, invalid args, ...) arrives on a perfectly healthy
                // channel; evicting it just churns the connection and pushes the peer toward
                // the offline threshold for no benefit. See issue #4567.
                let transport_failure = Self::is_transport_failure(&err);
                if Self::is_scanner_leader_lock(resource_summary) {
                    debug!(
                        addr = %self.addr,
                        op,
                        timeout_ms = lock_timeout.as_millis(),
                        resource_summary,
                        tonic_code = ?err.code(),
                        tonic_message = err.message(),
                        transport_failure,
                        "Remote lock RPC returned tonic error for scanner leader lock"
                    );
                } else {
                    warn!(
                        addr = %self.addr,
                        op,
                        timeout_ms = lock_timeout.as_millis(),
                        resource_summary,
                        tonic_code = ?err.code(),
                        tonic_message = err.message(),
                        transport_failure,
                        "Remote lock RPC returned tonic error"
                    );
                }
                if transport_failure {
                    self.evict_connection(op, &reason, resource_summary).await;
                }
                Err(LockError::internal(format!("{op} RPC failed: {reason}")))
            }
            Err(_) => {
                let reason = format!("RPC timed out after {:?}", lock_timeout);
                if Self::is_scanner_leader_lock(resource_summary) {
                    debug!(
                        addr = %self.addr,
                        op,
                        timeout_ms = lock_timeout.as_millis(),
                        resource_summary,
                        "Remote lock RPC timed out for scanner leader lock"
                    );
                } else {
                    warn!(
                        addr = %self.addr,
                        op,
                        timeout_ms = lock_timeout.as_millis(),
                        resource_summary,
                        "Remote lock RPC timed out"
                    );
                }
                self.evict_connection(op, &reason, resource_summary).await;
                Err(LockError::timeout(format!("remote lock RPC {op} on {}", self.addr), lock_timeout))
            }
        }
    }

    fn rpc_timeout_failure_response(request: &LockRequest, err: &LockError) -> LockResponse {
        LockResponse::failure(format!("Remote lock RPC timed out: {err}"), request.acquire_timeout)
    }

    fn rpc_failure_response(_request: &LockRequest, err: &LockError) -> LockResponse {
        LockResponse::failure(format!("Remote lock RPC failed: {err}"), Duration::ZERO)
    }

    fn rpc_failure_batch(requests: &[LockRequest], err: &LockError) -> Vec<LockResponse> {
        requests
            .iter()
            .map(|request| Self::rpc_failure_response(request, err))
            .collect()
    }

    fn rpc_timeout_failure_batch(requests: &[LockRequest], err: &LockError) -> Vec<LockResponse> {
        requests
            .iter()
            .map(|request| Self::rpc_timeout_failure_response(request, err))
            .collect()
    }

    fn build_lock_info(request: &LockRequest, lock_info_json: Option<String>) -> LockInfo {
        if let Some(lock_info_json) = lock_info_json {
            match serde_json::from_str::<LockInfo>(&lock_info_json) {
                Ok(info) => info,
                Err(e) => {
                    warn!("Failed to deserialize lock_info from response: {}, using request data", e);
                    LockInfo {
                        id: request.lock_id.clone(),
                        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,
                    }
                }
            }
        } else {
            LockInfo {
                id: request.lock_id.clone(),
                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,
            }
        }
    }

    fn unknown_lock_info(lock_id: &LockId) -> LockInfo {
        LockInfo {
            id: lock_id.clone(),
            resource: lock_id.resource.clone(),
            lock_type: LockType::Exclusive,
            status: LockStatus::Acquired,
            owner: "unknown".to_string(),
            acquired_at: std::time::SystemTime::now(),
            expires_at: std::time::SystemTime::now() + std::time::Duration::from_secs(3600),
            last_refreshed: std::time::SystemTime::now(),
            metadata: LockMetadata::default(),
            priority: LockPriority::Normal,
            wait_start_time: None,
        }
    }
}

#[async_trait]
impl LockClient for RemoteClient {
    async fn acquire_lock(&self, request: &LockRequest) -> Result<LockResponse> {
        info!("remote acquire_exclusive for {}", request.resource);
        let mut client = self.get_client().await?;
        let resource_summary = request.resource.to_string();
        let mut req = Request::new(GenerallyLockRequest {
            args: serde_json::to_string(&request)
                .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
        });
        attach_lock_mutation_body_digest(&mut req)?;

        let resp = match self.execute_rpc("lock", &resource_summary, client.lock(req)).await {
            Ok(resp) => resp.into_inner(),
            Err(err @ LockError::Timeout { .. }) => return Ok(Self::rpc_timeout_failure_response(request, &err)),
            Err(err) => return Ok(Self::rpc_failure_response(request, &err)),
        };

        // Check if the lock acquisition was successful
        if resp.success {
            Ok(LockResponse::success(
                Self::build_lock_info(request, resp.lock_info),
                std::time::Duration::ZERO,
            ))
        } else {
            // Lock acquisition failed
            Ok(LockResponse::failure(
                resp.error_info
                    .unwrap_or_else(|| "Lock acquisition failed on remote server".to_string()),
                std::time::Duration::ZERO,
            ))
        }
    }

    async fn acquire_locks_batch(&self, requests: &[LockRequest]) -> Result<Vec<LockResponse>> {
        if requests.is_empty() {
            return Ok(Vec::new());
        }

        let mut client = self.get_client().await?;
        let resource_summary = Self::summarize_resources(requests);
        let mut req = Request::new(BatchGenerallyLockRequest {
            args: requests
                .iter()
                .map(|request| {
                    serde_json::to_string(request).map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))
                })
                .collect::<Result<Vec<_>>>()?,
        });
        attach_lock_mutation_body_digest(&mut req)?;

        let resp = match self
            .execute_rpc("lock_batch", &resource_summary, client.lock_batch(req))
            .await
        {
            Ok(resp) => resp.into_inner(),
            Err(err @ LockError::Timeout { .. }) => return Ok(Self::rpc_timeout_failure_batch(requests, &err)),
            Err(err) => return Ok(Self::rpc_failure_batch(requests, &err)),
        };

        Ok(requests
            .iter()
            .enumerate()
            .map(|(idx, request)| match resp.results.get(idx) {
                Some(result) if result.success => {
                    LockResponse::success(Self::build_lock_info(request, result.lock_info.clone()), std::time::Duration::ZERO)
                }
                Some(result) => LockResponse::failure(
                    result
                        .error_info
                        .clone()
                        .unwrap_or_else(|| "Lock acquisition failed on remote server".to_string()),
                    std::time::Duration::ZERO,
                ),
                None => LockResponse::failure(
                    format!("Lock batch response missing entry for request index {idx}"),
                    std::time::Duration::ZERO,
                ),
            })
            .collect())
    }

    async fn release(&self, lock_id: &LockId) -> Result<bool> {
        info!("remote release for {}", lock_id);

        let unlock_request = Self::create_unlock_request(lock_id);
        let request_string = serde_json::to_string(&unlock_request)
            .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?;
        let mut client = self.get_client().await?;
        let resource_summary = unlock_request.resource.to_string();
        let mut req = Request::new(GenerallyLockRequest { args: request_string });
        attach_lock_mutation_body_digest(&mut req)?;
        let resp = self
            .execute_rpc("release", &resource_summary, client.un_lock(req))
            .await?
            .into_inner();
        if let Some(error_info) = resp.error_info {
            return Err(LockError::internal(error_info));
        }
        Ok(resp.success)
    }

    async fn release_locks_batch(&self, lock_ids: &[LockId]) -> Result<Vec<bool>> {
        if lock_ids.is_empty() {
            return Ok(Vec::new());
        }

        let unlock_requests = lock_ids.iter().map(Self::create_unlock_request).collect::<Vec<_>>();
        let mut client = self.get_client().await?;
        let resource_summary = Self::summarize_resources(&unlock_requests);
        let mut req = Request::new(BatchGenerallyLockRequest {
            args: unlock_requests
                .iter()
                .map(|request| {
                    serde_json::to_string(request).map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))
                })
                .collect::<Result<Vec<_>>>()?,
        });
        attach_lock_mutation_body_digest(&mut req)?;

        let resp = self
            .execute_rpc("release_batch", &resource_summary, client.un_lock_batch(req))
            .await?
            .into_inner();

        Ok(lock_ids
            .iter()
            .enumerate()
            .map(|(idx, _)| resp.results.get(idx).map(|result| result.success).unwrap_or(false))
            .collect())
    }

    async fn refresh(&self, lock_id: &LockId) -> Result<bool> {
        info!("remote refresh for {}", lock_id);
        let refresh_request = Self::create_unlock_request(lock_id);
        let mut client = self.get_client().await?;
        let resource_summary = refresh_request.resource.to_string();
        let mut req = Request::new(GenerallyLockRequest {
            args: serde_json::to_string(&refresh_request)
                .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
        });
        attach_lock_mutation_body_digest(&mut req)?;
        let resp = self
            .execute_rpc("refresh", &resource_summary, client.refresh(req))
            .await?
            .into_inner();
        if let Some(error_info) = resp.error_info {
            return Err(LockError::internal(error_info));
        }
        Ok(resp.success)
    }

    async fn force_release(&self, lock_id: &LockId) -> Result<bool> {
        info!("remote force_release for {}", lock_id);
        let force_request = Self::create_unlock_request(lock_id);
        let mut client = self.get_client().await?;
        let resource_summary = force_request.resource.to_string();
        let mut req = Request::new(GenerallyLockRequest {
            args: serde_json::to_string(&force_request)
                .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
        });
        attach_lock_mutation_body_digest(&mut req)?;
        let resp = self
            .execute_rpc("force_release", &resource_summary, client.force_un_lock(req))
            .await?
            .into_inner();
        if let Some(error_info) = resp.error_info {
            return Err(LockError::internal(error_info));
        }
        Ok(resp.success)
    }

    async fn check_status(&self, lock_id: &LockId) -> Result<Option<LockInfo>> {
        info!("remote check_status for {}", lock_id);

        // Since there's no direct status query in the gRPC service,
        // we attempt a non-blocking lock acquisition to check if the resource is available
        let status_request = Self::create_unlock_request(lock_id);
        let resource_summary = status_request.resource.to_string();
        let mut client = self.get_client().await?;

        // Try to acquire a very short-lived lock to test availability
        let mut req = Request::new(GenerallyLockRequest {
            args: serde_json::to_string(&status_request)
                .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
        });
        attach_lock_mutation_body_digest(&mut req)?;

        // Try exclusive lock first with very short timeout
        let resp = match self.execute_rpc("check_status", &resource_summary, client.lock(req)).await {
            Ok(response) => response.into_inner(),
            Err(_) => return Ok(Some(Self::unknown_lock_info(lock_id))),
        };

        if resp.success {
            // If we successfully acquired the lock, the resource was free.
            // Immediately release it on a best-effort basis.
            let mut release_req = Request::new(GenerallyLockRequest {
                args: serde_json::to_string(&status_request)
                    .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
            });
            attach_lock_mutation_body_digest(&mut release_req)?;
            let _ = self
                .execute_rpc("check_status_release", &resource_summary, client.un_lock(release_req))
                .await;

            Ok(None)
        } else {
            // Lock acquisition failed, meaning someone is holding it.
            // We can't determine the exact details remotely, so return a generic status.
            Ok(Some(Self::unknown_lock_info(lock_id)))
        }
    }

    async fn get_stats(&self) -> Result<LockStats> {
        info!("remote get_stats from {}", self.addr);

        // Since there's no direct statistics endpoint in the gRPC service,
        // we return basic stats indicating this is a remote client
        let stats = LockStats {
            last_updated: std::time::SystemTime::now(),
            ..Default::default()
        };

        // We could potentially enhance this by:
        // 1. Keeping local counters of operations performed
        // 2. Adding a stats gRPC method to the service
        // 3. Querying server health endpoints

        // For now, return minimal stats indicating remote connectivity
        Ok(stats)
    }

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

    async fn is_online(&self) -> bool {
        // Use Ping interface to test if remote service is online
        let mut client = match self.get_client().await {
            Ok(client) => client,
            Err(_) => {
                info!("remote client {} connection failed", self.addr);
                return false;
            }
        };

        let ping_req = Request::new(Self::build_ping_request());

        match client.ping(ping_req).await {
            Ok(_) => {
                info!("remote client {} is online", self.addr);
                true
            }
            Err(_) => {
                info!("remote client {} ping failed", self.addr);
                false
            }
        }
    }

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

#[cfg(test)]
mod tests {
    use super::*;
    use crate::runtime::sources as runtime_sources;
    use rustfs_lock::{ObjectKey, types::LockPriority};
    use tokio::net::TcpListener;
    use tokio::task::JoinHandle;
    use tonic::transport::Endpoint as TonicEndpoint;

    async fn spawn_hanging_listener() -> Option<(String, JoinHandle<()>)> {
        let listener = match TcpListener::bind("127.0.0.1:0").await {
            Ok(listener) => listener,
            Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return None,
            Err(err) => panic!("test listener should bind: {err}"),
        };
        let addr = format!("http://{}", listener.local_addr().expect("listener local address should be available"));
        let task = tokio::spawn(async move {
            if let Ok((stream, _)) = listener.accept().await {
                let _stream = stream;
                tokio::time::sleep(Duration::from_secs(2)).await;
            }
        });
        Some((addr, task))
    }

    async fn closed_listener_addr() -> Option<String> {
        let listener = match TcpListener::bind("127.0.0.1:0").await {
            Ok(listener) => listener,
            Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return None,
            Err(err) => panic!("test listener should bind: {err}"),
        };
        let addr = format!("http://{}", listener.local_addr().expect("listener local address should be available"));
        drop(listener);
        Some(addr)
    }

    async fn cache_lazy_channel(addr: &str) {
        let channel = TonicEndpoint::from_shared(addr.to_string()).unwrap().connect_lazy();
        runtime_sources::cache_test_node_channel(addr.to_string(), channel).await;
    }

    fn ensure_test_rpc_secret() {
        runtime_sources::ensure_test_rpc_secret();
    }

    fn test_lock_request(timeout_duration: Duration) -> LockRequest {
        LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner-a")
            .with_acquire_timeout(timeout_duration)
            .with_priority(LockPriority::Normal)
    }

    #[test]
    fn lock_mutation_helper_marks_single_and_batch_requests_for_rolling_auth() {
        let mut single = Request::new(GenerallyLockRequest {
            args: "single-lock".to_string(),
        });
        attach_lock_mutation_body_digest(&mut single).expect("single lock digest must be attached");
        assert!(
            single
                .extensions()
                .get::<crate::cluster::rpc::http_auth::RollingMutationBodyDigest>()
                .is_some()
        );

        let mut batch = Request::new(BatchGenerallyLockRequest {
            args: vec!["batch-lock".to_string()],
        });
        attach_lock_mutation_body_digest(&mut batch).expect("batch lock digest must be attached");
        assert!(
            batch
                .extensions()
                .get::<crate::cluster::rpc::http_auth::RollingMutationBodyDigest>()
                .is_some()
        );
    }

    #[tokio::test]
    #[serial_test::serial]
    async fn test_remote_client_acquire_lock_uses_rpc_timeout_and_evicts_connection() {
        ensure_test_rpc_secret();
        let Some((addr, accept_task)) = spawn_hanging_listener().await else {
            return;
        };
        cache_lazy_channel(&addr).await;
        assert!(runtime_sources::test_node_channel_is_cached(&addr).await);

        temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("50"))], async {
            let client = RemoteClient::new(addr.clone());
            let request = test_lock_request(Duration::from_millis(5));
            let started_at = tokio::time::Instant::now();

            let response = client.acquire_lock(&request).await.unwrap();
            let elapsed = started_at.elapsed();

            assert!(
                elapsed >= Duration::from_millis(40),
                "remote lock RPC should use configured transport timeout, got {elapsed:?}"
            );
            assert!(
                elapsed < Duration::from_secs(1),
                "test RPC timeout should keep the test fast, got {elapsed:?}"
            );
            assert!(!response.success, "timed out lock acquisition should fail");
            assert!(
                response
                    .error
                    .as_deref()
                    .is_some_and(|error| error.contains("Remote lock RPC timed out")),
                "expected remote RPC timeout marker, got {:?}",
                response.error
            );
            assert!(
                !runtime_sources::test_node_channel_is_cached(&addr).await,
                "transport timeout should evict cached connection"
            );
        })
        .await;

        accept_task.abort();
    }

    #[tokio::test]
    #[serial_test::serial]
    async fn test_remote_client_acquire_locks_batch_uses_rpc_timeout_and_evicts_connection() {
        ensure_test_rpc_secret();
        let Some((addr, accept_task)) = spawn_hanging_listener().await else {
            return;
        };
        cache_lazy_channel(&addr).await;
        assert!(runtime_sources::test_node_channel_is_cached(&addr).await);

        temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("50"))], async {
            let client = RemoteClient::new(addr.clone());
            let requests = vec![test_lock_request(Duration::from_millis(5))];
            let started_at = tokio::time::Instant::now();

            let responses = client.acquire_locks_batch(&requests).await.unwrap();
            let elapsed = started_at.elapsed();

            assert!(
                elapsed >= Duration::from_millis(40),
                "remote batch lock RPC should use configured transport timeout, got {elapsed:?}"
            );
            assert!(
                elapsed < Duration::from_secs(1),
                "test RPC timeout should keep the test fast, got {elapsed:?}"
            );
            assert_eq!(responses.len(), 1);
            assert!(!responses[0].success, "timed out batch lock acquisition should fail");
            assert!(
                responses[0]
                    .error
                    .as_deref()
                    .is_some_and(|error| error.contains("Remote lock RPC timed out")),
                "expected remote RPC timeout marker, got {:?}",
                responses[0].error
            );
            assert!(
                !runtime_sources::test_node_channel_is_cached(&addr).await,
                "batch transport timeout should evict cached connection"
            );
        })
        .await;

        accept_task.abort();
    }

    #[tokio::test]
    #[serial_test::serial]
    async fn test_remote_client_release_uses_rpc_timeout_and_evicts_connection() {
        ensure_test_rpc_secret();
        let Some((addr, accept_task)) = spawn_hanging_listener().await else {
            return;
        };
        cache_lazy_channel(&addr).await;
        assert!(runtime_sources::test_node_channel_is_cached(&addr).await);

        temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("50"))], async {
            let client = RemoteClient::new(addr.clone());
            let request = test_lock_request(Duration::from_millis(5));
            let started_at = tokio::time::Instant::now();

            let err = client.release(&request.lock_id).await.expect_err("release should time out");
            let elapsed = started_at.elapsed();

            assert!(
                elapsed >= Duration::from_millis(40),
                "remote release RPC should use configured transport timeout, got {elapsed:?}"
            );
            assert!(
                elapsed < Duration::from_secs(1),
                "test RPC timeout should keep the test fast, got {elapsed:?}"
            );
            assert!(matches!(err, LockError::Timeout { .. }), "expected remote release timeout, got {err:?}");
            assert!(
                !runtime_sources::test_node_channel_is_cached(&addr).await,
                "release timeout should evict cached connection"
            );
        })
        .await;

        accept_task.abort();
    }

    #[tokio::test]
    #[serial_test::serial]
    async fn test_remote_client_refresh_tonic_error_evicts_connection() {
        ensure_test_rpc_secret();
        let Some(addr) = closed_listener_addr().await else {
            return;
        };
        cache_lazy_channel(&addr).await;
        assert!(runtime_sources::test_node_channel_is_cached(&addr).await);

        temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("500"))], async {
            let client = RemoteClient::new(addr.clone());
            let request = test_lock_request(Duration::from_millis(5));

            let err = client
                .refresh(&request.lock_id)
                .await
                .expect_err("refresh should report tonic failure");

            assert!(
                err.to_string().contains("refresh RPC failed"),
                "expected refresh RPC failure marker, got {err}"
            );
            assert!(
                !runtime_sources::test_node_channel_is_cached(&addr).await,
                "refresh tonic error should evict cached connection"
            );
        })
        .await;
    }

    #[tokio::test]
    #[serial_test::serial]
    async fn test_remote_client_check_status_timeout_evicts_connection_and_preserves_status_shape() {
        ensure_test_rpc_secret();
        let Some((addr, accept_task)) = spawn_hanging_listener().await else {
            return;
        };
        cache_lazy_channel(&addr).await;
        assert!(runtime_sources::test_node_channel_is_cached(&addr).await);

        temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("50"))], async {
            let client = RemoteClient::new(addr.clone());
            let request = test_lock_request(Duration::from_millis(5));
            let started_at = tokio::time::Instant::now();

            let status = client.check_status(&request.lock_id).await.unwrap();
            let elapsed = started_at.elapsed();

            assert!(
                elapsed >= Duration::from_millis(40),
                "remote check_status RPC should use configured transport timeout, got {elapsed:?}"
            );
            assert!(
                elapsed < Duration::from_secs(1),
                "test RPC timeout should keep the test fast, got {elapsed:?}"
            );
            let info = status.expect("communication failure should preserve unknown lock status shape");
            assert_eq!(info.id, request.lock_id);
            assert_eq!(info.owner, "unknown");
            assert!(
                !runtime_sources::test_node_channel_is_cached(&addr).await,
                "check_status timeout should evict cached connection"
            );
        })
        .await;

        accept_task.abort();
    }

    #[test]
    fn test_is_transport_failure_classifies_only_channel_level_errors() {
        use tonic::{Code, Status};

        // Genuine transport failures: broken/unusable channel.
        for code in [Code::Unavailable, Code::DeadlineExceeded, Code::Unknown, Code::Cancelled] {
            assert!(
                RemoteClient::is_transport_failure(&Status::new(code, "boom")),
                "{code:?} should be treated as a transport failure"
            );
        }

        // A status carrying an underlying transport error as its source is a transport failure
        // regardless of code (tonic reports connection/h2 breakage this way).
        let sourced = Status::from_error(Box::new(std::io::Error::new(std::io::ErrorKind::ConnectionReset, "reset")));
        assert!(
            RemoteClient::is_transport_failure(&sourced),
            "a status with an underlying transport source should be a transport failure"
        );

        // Server-produced application statuses arrive on a healthy channel and must NOT evict it.
        for code in [
            Code::Unauthenticated,
            Code::PermissionDenied,
            Code::Internal,
            Code::FailedPrecondition,
            Code::InvalidArgument,
            Code::NotFound,
            Code::AlreadyExists,
            Code::ResourceExhausted,
            Code::Unimplemented,
            Code::Aborted,
            Code::OutOfRange,
        ] {
            assert!(
                !RemoteClient::is_transport_failure(&Status::new(code, "denied")),
                "{code:?} is an application status and must not be treated as a transport failure"
            );
        }
    }

    #[test]
    #[serial_test::serial]
    fn test_remote_client_rpc_timeout_honors_configured_deadline() {
        temp_env::with_var(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, None::<&str>, || {
            assert_eq!(
                RemoteClient::rpc_timeout(),
                Duration::from_millis(rustfs_config::DEFAULT_OBJECT_LOCK_RPC_TIMEOUT_MS)
            );
        });
        temp_env::with_var(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("50"), || {
            assert_eq!(RemoteClient::rpc_timeout(), Duration::from_millis(50));
        });
        temp_env::with_var(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("0"), || {
            assert_eq!(RemoteClient::rpc_timeout(), Duration::from_millis(1));
        });
    }
}