fxcp_core/nfs/

client.rs

// SPDX-License-Identifier: GPL-2.0-or-later
// Copyright (C) 2025 Joel Wirāmu Pauling <aenertia@aenertia.net>
//
// fxcp-core/src/nfs/client.rs  --  NFSv4.2 compound RPC TCP client

//! Persistent TCP client for sending NFSv4.2 compound RPCs directly to the
//! NFS server, bypassing the Linux VFS for small-file writes.
//!
//! Manages session establishment (EXCHANGE_ID + CREATE_SESSION), sequence IDs,
//! and directory handle caching.

#![allow(clippy::unwrap_used)]
use std::io::{Read, Write};
use std::net::TcpStream;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering};
use dashmap::DashMap;
use tracing::{debug, info, warn};

use crate::constants;

use super::mount::NfsBypassInfo;
use super::rpc::{self, Nfs4Op, StateId, WriteStable};
use super::NfsError;

/// Transport abstraction for NFS compound RPCs.
///
/// Without the `tls` or `rdma` features: zero-cost newtype wrapping `TcpStream`.
/// With the `tls` feature: enum supporting plaintext TCP or TLS 1.3.
/// With the `rdma` feature: enum supporting plaintext TCP or RDMA Send/Recv.
/// All variants implement `Read + Write` so stream-based I/O sites are unchanged.
/// The RDMA variant bypasses Read/Write  --  it uses `rpc_call()` directly.
#[cfg(not(any(feature = "tls", feature = "rdma")))]
pub(crate) struct NfsTransport(TcpStream);

#[cfg(not(any(feature = "tls", feature = "rdma")))]
impl NfsTransport {
    /// Wrap a TcpStream in NfsTransport (zero cost without tls feature).
    pub fn new(s: TcpStream) -> Self { Self(s) }
    /// Borrow the underlying TcpStream for socket option calls.
    pub fn tcp_ref(&self) -> &TcpStream { &self.0 }
    /// Apply standard TCP options: no-delay, read/write timeouts.
    pub fn configure_tcp(&self) -> std::io::Result<()> {
        use std::time::Duration;
        self.0.set_nodelay(true)?;
        self.0.set_read_timeout(Some(Duration::from_secs(constants::NFS_TCP_RW_TIMEOUT_SECS)))?;
        self.0.set_write_timeout(Some(Duration::from_secs(constants::NFS_TCP_RW_TIMEOUT_SECS)))?;
        Ok(())
    }
}

#[cfg(not(any(feature = "tls", feature = "rdma")))]
impl std::io::Read for NfsTransport {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> { self.0.read(buf) }
}

#[cfg(not(any(feature = "tls", feature = "rdma")))]
impl std::io::Write for NfsTransport {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> { self.0.write(buf) }
    fn flush(&mut self) -> std::io::Result<()> { self.0.flush() }
}

/// RDMA transport (without TLS  --  mutually exclusive per RFC 9289 S5.1.3).
///
/// Enum with TCP fallback for session establishment (EXCHANGE_ID, CREATE_SESSION)
/// and RDMA for data-path RPC compounds. TCP is used initially; the RDMA variant
/// is selected when `proto=rdma` is detected in mount options (Wave 5).
#[cfg(all(feature = "rdma", not(feature = "tls")))]
pub(crate) enum NfsTransport {
    /// Plaintext TCP connection (default, used for session establishment).
    Tcp(TcpStream),
    /// RDMA Send/Recv connection (RFC 8166). Does NOT implement Read/Write  -- 
    /// all I/O goes through `NfsCompoundClient::rpc_call()`.
    Rdma(super::rdma_transport::RdmaConnection),
}

#[cfg(all(feature = "rdma", not(feature = "tls")))]
impl NfsTransport {
    /// Wrap a TcpStream as a plaintext transport.
    pub fn new(s: TcpStream) -> Self { NfsTransport::Tcp(s) }
    /// Borrow the underlying TcpStream for socket option calls.
    ///
    /// # Panics
    ///
    /// Panics if called on an `NfsTransport::Rdma`  --  RDMA has no TCP socket.
    pub fn tcp_ref(&self) -> &TcpStream {
        match self {
            NfsTransport::Tcp(s) => s,
            NfsTransport::Rdma(_) => unreachable!("RDMA transport has no TCP socket"),
        }
    }
    /// Apply standard TCP options to the inner TcpStream.
    pub fn configure_tcp(&self) -> std::io::Result<()> {
        use std::time::Duration;
        let tcp = self.tcp_ref();
        tcp.set_nodelay(true)?;
        tcp.set_read_timeout(Some(Duration::from_secs(constants::NFS_TCP_RW_TIMEOUT_SECS)))?;
        tcp.set_write_timeout(Some(Duration::from_secs(constants::NFS_TCP_RW_TIMEOUT_SECS)))?;
        Ok(())
    }
}

#[cfg(all(feature = "rdma", not(feature = "tls")))]
impl std::io::Read for NfsTransport {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        match self {
            NfsTransport::Tcp(s) => s.read(buf),
            NfsTransport::Rdma(_) => unreachable!("RDMA uses rpc_call() directly, not Read"),
        }
    }
}

#[cfg(all(feature = "rdma", not(feature = "tls")))]
impl std::io::Write for NfsTransport {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        match self {
            NfsTransport::Tcp(s) => s.write(buf),
            NfsTransport::Rdma(_) => unreachable!("RDMA uses rpc_call() directly, not Write"),
        }
    }
    fn flush(&mut self) -> std::io::Result<()> {
        match self {
            NfsTransport::Tcp(s) => s.flush(),
            NfsTransport::Rdma(_) => unreachable!("RDMA uses rpc_call() directly, not Write"),
        }
    }
}

#[cfg(feature = "tls")]
pub(crate) enum NfsTransport {
    /// Plaintext TCP connection (default / pre-TLS-upgrade).
    Tcp(TcpStream),
    /// TLS 1.3 connection established via RFC 9289 STARTTLS upgrade.
    Tls(Box<rustls::StreamOwned<rustls::ClientConnection, TcpStream>>),
    /// kTLS-promoted socket  --  kernel handles AES-GCM transparently.
    /// After promotion the rustls `ClientConnection` is consumed (secrets
    /// extracted via `dangerous_extract_secrets`), so I/O goes directly
    /// through the `TcpStream` fd where the kernel TLS ULP encrypts/decrypts.
    #[cfg(feature = "ktls")]
    Ktls(TcpStream),
    /// RDMA Send/Recv connection (RFC 8166). Mutually exclusive with TLS
    /// per RFC 9289 S5.1.3  --  use krb5p for encryption over RDMA.
    #[cfg(feature = "rdma")]
    Rdma(super::rdma_transport::RdmaConnection),
}

#[cfg(feature = "tls")]
impl NfsTransport {
    /// Wrap a TcpStream as a plaintext transport.
    pub fn new(s: TcpStream) -> Self { NfsTransport::Tcp(s) }
    /// Borrow the underlying TcpStream for socket option calls.
    pub fn tcp_ref(&self) -> &TcpStream {
        match self {
            NfsTransport::Tcp(s) => s,
            NfsTransport::Tls(s) => s.get_ref(),
            #[cfg(feature = "ktls")]
            NfsTransport::Ktls(s) => s,
            #[cfg(feature = "rdma")]
            NfsTransport::Rdma(_) => unreachable!("RDMA transport has no TCP socket"),
        }
    }
    /// Apply standard TCP options to the inner TcpStream.
    pub fn configure_tcp(&self) -> std::io::Result<()> {
        use std::time::Duration;
        let tcp = self.tcp_ref();
        tcp.set_nodelay(true)?;
        tcp.set_read_timeout(Some(Duration::from_secs(constants::NFS_TCP_RW_TIMEOUT_SECS)))?;
        tcp.set_write_timeout(Some(Duration::from_secs(constants::NFS_TCP_RW_TIMEOUT_SECS)))?;
        Ok(())
    }
}

#[cfg(feature = "tls")]
impl std::io::Read for NfsTransport {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        match self {
            NfsTransport::Tcp(s) => s.read(buf),
            NfsTransport::Tls(s) => s.read(buf),
            #[cfg(feature = "ktls")]
            NfsTransport::Ktls(s) => s.read(buf),
            #[cfg(feature = "rdma")]
            NfsTransport::Rdma(_) => unreachable!("RDMA uses rpc_call() directly"),
        }
    }
}

#[cfg(feature = "tls")]
impl std::io::Write for NfsTransport {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        match self {
            NfsTransport::Tcp(s) => s.write(buf),
            NfsTransport::Tls(s) => s.write(buf),
            #[cfg(feature = "ktls")]
            NfsTransport::Ktls(s) => s.write(buf),
            #[cfg(feature = "rdma")]
            NfsTransport::Rdma(_) => unreachable!("RDMA uses rpc_call() directly"),
        }
    }
    fn flush(&mut self) -> std::io::Result<()> {
        match self {
            NfsTransport::Tcp(s) => s.flush(),
            NfsTransport::Tls(s) => s.flush(),
            #[cfg(feature = "ktls")]
            NfsTransport::Ktls(s) => s.flush(),
            #[cfg(feature = "rdma")]
            NfsTransport::Rdma(_) => unreachable!("RDMA uses rpc_call() directly"),
        }
    }
}

#[cfg(target_os = "linux")]
impl NfsTransport {
    /// Raw file descriptor of the underlying TCP socket (for setsockopt).
    #[allow(dead_code)]
    pub fn as_raw_fd(&self) -> std::os::unix::io::RawFd {
        use std::os::unix::io::AsRawFd;
        self.tcp_ref().as_raw_fd()
    }
}

#[cfg(feature = "tls")]
impl NfsTransport {
    /// Export TLS keying material for RFC 9266 tls-exporter channel binding.
    ///
    /// Returns 32-byte EKM when the transport is TLS-protected, `None` for plaintext.
    /// The EKM label is `EXPORTER-Channel-Binding` per RFC 9266 S3.
    pub fn export_channel_binding(&self) -> Option<Vec<u8>> {
        match self {
            NfsTransport::Tls(tls) => {
                let mut ekm = [0u8; 32];
                match tls.conn.export_keying_material(&mut ekm, b"EXPORTER-Channel-Binding", None) {
                    Ok(_) => Some(ekm.to_vec()),
                    Err(e) => {
                        tracing::warn!("TLS EKM export failed (channel binding unavailable): {e}");
                        None
                    }
                }
            }
            NfsTransport::Tcp(_) => None,
            #[cfg(feature = "ktls")]
            NfsTransport::Ktls(_) => None,
            #[cfg(feature = "rdma")]
            NfsTransport::Rdma(_) => None,
        }
    }
}

/// AIMD self-tuner for NFSv4 compound RPC operation count.
///
/// Increases files-per-compound after sustained success (additive increase)
/// and halves on server rejection (multiplicative decrease).
/// Derived from server's negotiated ca_maxoperations in CREATE_SESSION.
struct CompoundSizeTuner {
    /// Current target: files per compound [1..=negotiated_ceiling]
    current_max: AtomicUsize,
    /// Server's negotiated limit: (ca_maxoperations - 2) / 4
    negotiated_ceiling: usize,
    /// Consecutive successes at current_max (reset on increase or error)
    consecutive_success: AtomicU32,
}

impl CompoundSizeTuner {
    fn new(negotiated_max_ops: u32) -> Self {
        // (ca_maxops - 2 header ops) / 3 ops per file (OPEN+WRITE+CLOSE)
        let ceiling = ((negotiated_max_ops.saturating_sub(2)) / 3).max(1) as usize;
        // Start conservative at 3 (current baseline), grow to ceiling
        Self {
            current_max: AtomicUsize::new(ceiling.min(3)),
            negotiated_ceiling: ceiling,
            consecutive_success: AtomicU32::new(0),
        }
    }

    /// Current files-per-compound target.
    fn current(&self) -> usize {
        self.current_max.load(Ordering::Relaxed)
    }

    /// Called after a successful compound. Adds 1 after 10 consecutive successes.
    fn on_success(&self) {
        let succ = self.consecutive_success.fetch_add(1, Ordering::Relaxed) + 1;
        if succ >= 10 {
            let cur = self.current_max.load(Ordering::Relaxed);
            if cur < self.negotiated_ceiling {
                self.current_max.store(cur + 1, Ordering::Relaxed);
                self.consecutive_success.store(0, Ordering::Relaxed);
                debug!("CompoundSizeTuner: increased to {} files/compound", cur + 1);
            }
        }
    }

    /// Called on NFS4ERR_REQ_TOO_BIG (10019) or NFS4ERR_REP_TOO_BIG (10028).
    fn on_size_error(&self) {
        let cur = self.current_max.load(Ordering::Relaxed);
        self.current_max.store((cur / 2).max(1), Ordering::Relaxed);
        self.consecutive_success.store(0, Ordering::Relaxed);
        warn!("CompoundSizeTuner: halved to {} files/compound", (cur / 2).max(1));
    }
}

/// NFSv4.2 compound RPC client over TCP.
///
/// Sends file creation compounds directly to the NFS server, bypassing VFS.
/// One TCP connection, one session, sequential slot usage.
pub struct NfsCompoundClient {
    stream: NfsTransport,
    session_id: [u8; 16],
    sequence_id: AtomicU32,
    client_id: u64,
    uid: u32,
    gid: u32,
    machine: String,
    xid_counter: AtomicU32,
    server_info: NfsBypassInfo,
    /// Cache of directory path -> NFS file handle.
    dir_handle_cache: DashMap<PathBuf, Vec<u8>>,
    /// AIMD tuner for files-per-compound sizing.
    compound_tuner: CompoundSizeTuner,
    /// GSS security context for RPCSEC_GSS (krb5/krb5i/krb5p mounts).
    #[cfg(feature = "krb5")]
    gss_ctx: Option<super::gss::GssContext>,
}

/// Set TCP_QUICKACK to disable delayed ACKs for this send/recv cycle.
/// Must be called before each read_reply()  --  TCP_QUICKACK resets after each ACK.
#[cfg(target_os = "linux")]
fn set_quickack(stream: &NfsTransport) {
    use std::os::unix::io::AsRawFd;
    let val: libc::c_int = 1;
    // SAFETY: tcp_ref() returns a valid TCP socket fd. TCP_QUICKACK is a standard Linux socket option.
    let ret = unsafe {
        libc::setsockopt(
            stream.tcp_ref().as_raw_fd(),
            libc::IPPROTO_TCP,
            libc::TCP_QUICKACK,
            &val as *const _ as *const libc::c_void,
            std::mem::size_of_val(&val) as libc::socklen_t,
        )
    };
    if ret != 0 {
        tracing::debug!("TCP_QUICKACK setsockopt failed: {}", std::io::Error::last_os_error());
    }
}

#[cfg(not(target_os = "linux"))]
fn set_quickack(_stream: &NfsTransport) {}

impl NfsCompoundClient {
    /// Connect to the NFS server and establish a session.
    ///
    /// Performs TCP connect -> EXCHANGE_ID -> CREATE_SESSION.
    pub fn connect(info: &NfsBypassInfo) -> Result<Self, NfsError> {
        // NFS servers default to `secure`  --  require source port <1024.
        // Bind to a privileged port before connecting (requires root/CAP_NET_BIND_SERVICE).
        let tcp_stream = Self::connect_privileged(&info.server_addr)?;
        let stream = NfsTransport::new(tcp_stream);
        stream.configure_tcp()?;
        // Disable delayed ACKs for lower RPC latency
        set_quickack(&stream);

        // Use effective UID/GID directly. The NFS server handles root_squash
        // mapping  --  we send our real credentials and let the server decide.
        // SAFETY: getuid() is always safe and cannot fail.
        let uid = unsafe { libc::getuid() };
        // SAFETY: getgid() is always safe and cannot fail.
        let gid = unsafe { libc::getgid() };
        let machine = {
            let mut buf = [0u8; constants::NFS_HOSTNAME_BUFFER_SIZE];
            // SAFETY: buf is a valid stack buffer. gethostname writes
            // at most buf.len() bytes including the null terminator.
            if unsafe { libc::gethostname(buf.as_mut_ptr() as *mut _, buf.len()) } == 0 {
                let len = buf.iter().position(|&b| b == 0).unwrap_or(buf.len());
                String::from_utf8_lossy(&buf[..len]).to_string()
            } else {
                "foxing".to_string()
            }
        };

        let mut client = Self {
            stream,
            session_id: [0u8; 16],
            sequence_id: AtomicU32::new(1),
            client_id: 0,
            uid,
            gid,
            machine,
            xid_counter: AtomicU32::new(1),
            server_info: info.clone(),
            dir_handle_cache: DashMap::new(),
            compound_tuner: CompoundSizeTuner::new(crate::constants::NFS_COMPOUND_TUNER_INITIAL_MAX_OPS),
            #[cfg(feature = "krb5")]
            gss_ctx: None,
        };

        // TLS transport upgrade (RFC 9289 RPC-over-TLS) if xprtsec= mount option detected
        #[cfg(feature = "tls")]
        {
            use super::NfsTransportSecurity;
            match info.transport_security {
                NfsTransportSecurity::Tls => {
                    client.try_tls_upgrade()?;
                    info!("NFS bypass: TLS transport active (xprtsec=tls)");
                }
                NfsTransportSecurity::Opportunistic => {
                    match client.try_tls_upgrade() {
                        Ok(()) => info!("NFS bypass: TLS transport active (opportunistic)"),
                        Err(NfsError::TlsNotSupported) => {
                            warn!(
                                "NFS bypass: server {} does not support xprtsec=tls, \
                                 continuing with cleartext transport",
                                info.server_addr
                            );
                            crate::metrics::NFS_BYPASS_TLS_FALLBACK.inc();
                        }
                        Err(e) => {
                            warn!("NFS bypass: TLS opportunistic upgrade failed ({}), \
                                   continuing cleartext", e);
                            crate::metrics::NFS_BYPASS_TLS_FALLBACK.inc();
                        }
                    }
                }
                NfsTransportSecurity::None => {}
            }
        }

        // Double-encryption advisory: sec=krb5p + TLS = redundant AES layers
        #[cfg(all(feature = "tls", feature = "krb5"))]
        if info.security == super::NfsSecurity::Krb5p
            && info.transport_security != super::NfsTransportSecurity::None
        {
            warn!(
                "NFS bypass: double encryption detected  --  sec=krb5p (per-RPC AES-256) AND \
                 xprtsec=tls (TLS 1.3 AES-GCM) are both active. \
                 Consider sec=krb5,xprtsec=tls for ~5% overhead instead of ~30%. \
                 Both layers provide full confidentiality; combining adds no security gain."
            );
            crate::metrics::NFS_BYPASS_DOUBLE_ENCRYPTION.inc();
        }

        // Session bootstrap: EXCHANGE_ID -> CREATE_SESSION
        client.establish_session()?;

        // If mount requires Kerberos, attempt GSS context establishment.
        // On failure, log warning and continue with AUTH_SYS (server may reject).
        #[cfg(feature = "krb5")]
        if info.security.requires_kerberos() {
            // RFC 9266 tls-exporter channel binding: bind GSS context to TLS session
            #[cfg(feature = "tls")]
            let channel_binding = client.stream.export_channel_binding();
            #[cfg(not(feature = "tls"))]
            let channel_binding: Option<Vec<u8>> = None;

            let hostname = info.server_hostname.clone();
            match super::gss::GssContext::establish(
                &hostname,
                info.security,
                channel_binding.as_deref(),
            ) {
                Ok(mut gss) => {
                    match client.do_rpcsec_gss_init(&mut gss) {
                        Ok(()) => {
                            info!(
                                "NFS bypass: RPCSEC_GSS context ready (sec={}, server_handle={} bytes)",
                                info.security, gss.server_handle.len()
                            );
                            client.gss_ctx = Some(gss);
                        }
                        Err(e) => {
                            warn!(
                                "NFS bypass: RPCSEC_GSS INIT failed for {} (sec={}): {}  --  \
                                 falling back to AUTH_SYS",
                                info.server_addr, info.security, e
                            );
                        }
                    }
                }
                Err(e) => {
                    warn!(
                        "NFS bypass: GSS context failed for {} (sec={}): {}  --  \
                         falling back to AUTH_SYS (may be rejected by server)",
                        info.server_addr, info.security, e
                    );
                }
            }
        }

        info!("NFS bypass: session established with {} (client_id={:#x})",
              info.server_addr, client.client_id);

        Ok(client)
    }

    /// Establish NFSv4.1+ session via EXCHANGE_ID + CREATE_SESSION.
    fn establish_session(&mut self) -> Result<(), NfsError> {
        // Step 1: EXCHANGE_ID  --  register client identity with server
        let client_id = self.do_exchange_id()?;
        self.client_id = client_id;
        debug!("NFS bypass: EXCHANGE_ID ok, client_id={:#x}", client_id);

        // Step 2: CREATE_SESSION  --  get session_id and slot table
        let session_id = self.do_create_session(client_id)?;
        self.session_id = session_id;
        self.sequence_id.store(1, Ordering::Relaxed);
        debug!("NFS bypass: CREATE_SESSION ok, session={:02x?}", &session_id[..4]);

        // Step 3: RECLAIM_COMPLETE  --  tell server we have no state to reclaim.
        // This ends the grace period for our session so OPEN works immediately.
        self.do_reclaim_complete()?;
        debug!("NFS bypass: RECLAIM_COMPLETE ok");

        Ok(())
    }

    /// Send EXCHANGE_ID compound to get a client_id from the server.
    fn do_exchange_id(&mut self) -> Result<u64, NfsError> {
        let xid = self.next_xid();

        // Build EXCHANGE_ID compound (no SEQUENCE  --  it's a session-creating op)
        let mut body = super::xdr::XdrEncoder::new(constants::NFS_XDR_ENCODER_SMALL_CAPACITY);

        // RPC header
        body.encode_u32(xid);
        body.encode_u32(0); // CALL
        body.encode_u32(rpc::RPC_VERSION);
        body.encode_u32(rpc::NFS_PROGRAM);
        body.encode_u32(rpc::NFS_V4);
        body.encode_u32(rpc::NFSPROC4_COMPOUND);

        // AUTH_SYS
        encode_auth_sys(&mut body, self.uid, self.gid, &self.machine);
        // Verifier AUTH_NONE
        body.encode_u32(rpc::AUTH_NONE);
        body.encode_u32(0);

        // COMPOUND args: tag, minorversion=2, ops=[EXCHANGE_ID]
        body.encode_string("exid");
        body.encode_u32(2); // minorversion
        body.encode_u32(1); // 1 operation

        // EXCHANGE_ID operation
        body.encode_u32(rpc::OP_EXCHANGE_ID);
        // eia_clientowner: co_verifier(8) + co_ownerid(opaque)
        let verifier = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos() as u64;
        body.encode_u64(verifier); // co_verifier
        let owner_id = format!("foxing.{}.{}", self.machine, std::process::id());
        body.encode_opaque(owner_id.as_bytes()); // co_ownerid
        // eia_flags
        body.encode_u32(0x00000001); // EXCHGID4_FLAG_SUPP_MOVED_REFER
        // eia_state_protect: SP4_NONE
        body.encode_u32(0); // SP4_NONE
        // eia_client_impl_id: empty array
        body.encode_u32(0); // 0 elements

        let body_bytes = body.into_bytes();
        let reply_data = self.rpc_call(xid, &body_bytes)?;
        let reply = rpc::parse_compound_reply(&reply_data)?;

        if reply.status != rpc::NFS4_OK {
            return Err(NfsError::SessionFailed(format!(
                "EXCHANGE_ID failed: {} ({})", rpc::nfs4_error_name(reply.status), reply.status
            )));
        }

        let client_id = self.extract_exchange_id_client_id(&reply_data)?;
        Ok(client_id)
    }

    /// Extract client_id from EXCHANGE_ID reply by scanning for it in the raw data.
    fn extract_exchange_id_client_id(&self, data: &[u8]) -> Result<u64, NfsError> {
        // After the RPC reply header and compound header, find EXCHANGE_ID result.
        // The EXCHANGE_ID result starts after: op(4) + status(4), then:
        //   eir_clientid(8), eir_sequenceid(4), eir_flags(4), ...
        // We scan backwards from known structure. Simplified: look for the op code
        // OP_EXCHANGE_ID (42) followed by NFS4_OK (0), then read the next 8 bytes.
        let needle = [
            0, 0, 0, rpc::OP_EXCHANGE_ID as u8,  // op = 42
            0, 0, 0, 0,                            // status = 0 (NFS4_OK)
        ];
        if let Some(pos) = data.windows(8).position(|w| w == needle) {
            let cid_start = pos + 8;
            if cid_start + 8 <= data.len() {
                let client_id = u64::from_be_bytes([
                    data[cid_start], data[cid_start+1], data[cid_start+2], data[cid_start+3],
                    data[cid_start+4], data[cid_start+5], data[cid_start+6], data[cid_start+7],
                ]);
                return Ok(client_id);
            }
        }
        Err(NfsError::SessionFailed("cannot parse client_id from EXCHANGE_ID reply".into()))
    }

    /// Send CREATE_SESSION compound to get a session_id.
    fn do_create_session(&mut self, client_id: u64) -> Result<[u8; 16], NfsError> {
        let xid = self.next_xid();

        let mut body = super::xdr::XdrEncoder::new(constants::NFS_XDR_ENCODER_SMALL_CAPACITY);

        // RPC header
        body.encode_u32(xid);
        body.encode_u32(0); // CALL
        body.encode_u32(rpc::RPC_VERSION);
        body.encode_u32(rpc::NFS_PROGRAM);
        body.encode_u32(rpc::NFS_V4);
        body.encode_u32(rpc::NFSPROC4_COMPOUND);

        // AUTH_SYS
        encode_auth_sys(&mut body, self.uid, self.gid, &self.machine);
        body.encode_u32(rpc::AUTH_NONE);
        body.encode_u32(0);

        // COMPOUND args
        body.encode_string("csess");
        body.encode_u32(2); // minorversion
        body.encode_u32(1); // 1 operation

        // CREATE_SESSION operation (RFC 8881 S18.36)
        body.encode_u32(rpc::OP_CREATE_SESSION);
        body.encode_u64(client_id);          // csa_clientid
        body.encode_u32(1);                   // csa_sequence (from EXCHANGE_ID eir_sequenceid)
        body.encode_u32(0);                   // csa_flags: 0 (no persist, no back channel)
        // csa_fore_chan_attrs (channel_attrs4):
        //   headerpadsize, maxrequestsize, maxresponsesize,
        //   maxresponsesize_cached, maxoperations, maxrequests, rdma_ird[]
        body.encode_u32(0);                   // ca_headerpadsize
        body.encode_u32(crate::constants::NFS_MAX_REQUEST_SIZE as u32); // ca_maxrequestsize
        body.encode_u32(crate::constants::NFS_MAX_RESPONSE_SIZE as u32); // ca_maxresponsesize
        body.encode_u32(4096);                // ca_maxresponsesize_cached
        body.encode_u32(crate::constants::NFS_CA_MAX_OPERATIONS);  // ca_maxoperations  --  request higher limit; server negotiates down
        body.encode_u32(1);                   // ca_maxrequests (1 slot)
        body.encode_u32(0);                   // ca_rdma_ird count (empty array)
        // csa_back_chan_attrs (minimal  --  no back channel)
        body.encode_u32(0);                   // ca_headerpadsize
        body.encode_u32(4096);                // ca_maxrequestsize
        body.encode_u32(4096);                // ca_maxresponsesize
        body.encode_u32(0);                   // ca_maxresponsesize_cached
        body.encode_u32(2);                   // ca_maxoperations
        body.encode_u32(0);                   // ca_maxrequests (0 = no back channel)
        body.encode_u32(0);                   // ca_rdma_ird count
        // csa_cb_program
        body.encode_u32(0x40000000);          // callback program number (unused)
        // csa_sec_parms: callback_sec_parms4[]
        // For AUTH_SYS: secflavor(4) + authsys_parms
        body.encode_u32(1);                   // array count: 1 element
        body.encode_u32(rpc::AUTH_SYS);       // cb_secflavor
        // authsys_parms (cbsp_sys_cred):
        body.encode_u32(0);                   // stamp
        body.encode_string(&self.machine);    // machinename
        body.encode_u32(self.uid);            // uid
        body.encode_u32(self.gid);            // gid
        body.encode_u32(1);                   // gids count
        body.encode_u32(self.gid);            // gids[0]

        let body_bytes = body.into_bytes();
        let reply_data = self.rpc_call(xid, &body_bytes)?;
        let reply = rpc::parse_compound_reply(&reply_data)?;

        if reply.status != rpc::NFS4_OK {
            return Err(NfsError::SessionFailed(format!(
                "CREATE_SESSION failed: {} ({})", rpc::nfs4_error_name(reply.status), reply.status
            )));
        }

        // Extract session_id and negotiated ca_maxrequestsize from CREATE_SESSION reply
        let session_id = self.extract_session_id(&reply_data)?;

        // Log server's negotiated channel attributes for debugging size limits
        let needle = [
            0, 0, 0, rpc::OP_CREATE_SESSION as u8,
            0, 0, 0, 0,
        ];
        if let Some(pos) = reply_data.windows(8).position(|w| w == needle) {
            // CREATE_SESSION4resok: session_id(16) + sequence_id(4) + flags(4) + fore_chan_attrs
            // fore_chan_attrs: headerpadsize(4) + maxrequestsize(4) + maxresponsesize(4) + ...
            let attrs_start = pos + 8 + 16 + 4 + 4; // skip session_id + seq + flags
            if attrs_start + 12 <= reply_data.len() {
                let max_req = u32::from_be_bytes([
                    reply_data[attrs_start + 4], reply_data[attrs_start + 5],
                    reply_data[attrs_start + 6], reply_data[attrs_start + 7],
                ]);
                let max_ops = if attrs_start + 20 <= reply_data.len() {
                    u32::from_be_bytes([
                        reply_data[attrs_start + 16], reply_data[attrs_start + 17],
                        reply_data[attrs_start + 18], reply_data[attrs_start + 19],
                    ])
                } else { 0 };
                debug!(
                    "NFS bypass: server negotiated ca_maxrequestsize={} ca_maxoperations={}",
                    max_req, max_ops
                );
            }
        }

        Ok(session_id)
    }

    /// Extract session_id (16 bytes) from CREATE_SESSION reply.
    fn extract_session_id(&self, data: &[u8]) -> Result<[u8; 16], NfsError> {
        // CREATE_SESSION result starts after op(4) + status(4), then session_id(16)
        let needle = [
            0, 0, 0, rpc::OP_CREATE_SESSION as u8,
            0, 0, 0, 0,
        ];
        if let Some(pos) = data.windows(8).position(|w| w == needle) {
            let sid_start = pos + 8;
            if sid_start + 16 <= data.len() {
                let mut session_id = [0u8; 16];
                session_id.copy_from_slice(&data[sid_start..sid_start + 16]);
                return Ok(session_id);
            }
        }
        Err(NfsError::SessionFailed("cannot parse session_id from CREATE_SESSION reply".into()))
    }

    /// Send RECLAIM_COMPLETE to end the grace period for our session.
    fn do_reclaim_complete(&mut self) -> Result<(), NfsError> {
        let seq_id = self.next_sequence_id();
        let ops = vec![
            Nfs4Op::Sequence {
                session_id: self.session_id,
                sequence_id: seq_id,
                slot_id: 0,
                highest_slot_id: 0,
                cache_this: false,
            },
            Nfs4Op::ReclaimComplete,
        ];
        let xid = self.next_xid();
        let msg = self.build_compound_msg(xid, "rclm", &ops)?;
        let reply_data = self.rpc_call(xid, &msg)?;
        let reply = rpc::parse_compound_reply(&reply_data)?;
        if reply.status != rpc::NFS4_OK {
            return Err(NfsError::SessionFailed(format!(
                "RECLAIM_COMPLETE failed: {} ({})", rpc::nfs4_error_name(reply.status), reply.status
            )));
        }
        Ok(())
    }

    /// Re-establish session after NFS4ERR_BADSESSION or connection loss.
    pub fn recover_session(&mut self) -> Result<(), NfsError> {
        warn!("NFS bypass: recovering session...");
        // Reconnect TCP
        let new_tcp = TcpStream::connect_timeout(
            &self.server_info.server_addr,
            std::time::Duration::from_secs(constants::NFS_TCP_CONNECT_TIMEOUT_SECS),
        )?;
        self.stream = NfsTransport::new(new_tcp);
        self.stream.configure_tcp()?;
        set_quickack(&self.stream);

        // Re-establish TLS if the original connection used it
        #[cfg(feature = "tls")]
        {
            use super::NfsTransportSecurity;
            match self.server_info.transport_security {
                NfsTransportSecurity::Tls => {
                    self.try_tls_upgrade()?;
                    debug!("NFS recovery: TLS re-established");
                }
                NfsTransportSecurity::Opportunistic => {
                    match self.try_tls_upgrade() {
                        Ok(()) => debug!("NFS recovery: TLS re-established (opportunistic)"),
                        Err(e) => {
                            warn!("NFS recovery: TLS re-upgrade failed ({}), continuing cleartext", e);
                            crate::metrics::NFS_BYPASS_TLS_FALLBACK.inc();
                        }
                    }
                }
                NfsTransportSecurity::None => {}
            }
        }

        // Re-establish session
        self.establish_session()?;

        // Clear handle cache (handles may be stale after session loss)
        self.dir_handle_cache.clear();

        // Re-establish GSS context if Kerberos mount
        #[cfg(feature = "krb5")]
        if self.server_info.security.requires_kerberos() {
            #[cfg(feature = "tls")]
            let channel_binding = self.stream.export_channel_binding();
            #[cfg(not(feature = "tls"))]
            let channel_binding: Option<Vec<u8>> = None;

            let hostname = self.server_info.server_hostname.clone();
            match super::gss::GssContext::establish(
                &hostname,
                self.server_info.security,
                channel_binding.as_deref(),
            ) {
                Ok(mut gss) => {
                    match self.do_rpcsec_gss_init(&mut gss) {
                        Ok(()) => {
                            self.gss_ctx = Some(gss);
                            debug!("NFS bypass: GSS context re-established and INIT complete after recovery");
                        }
                        Err(e) => {
                            warn!("NFS bypass: RPCSEC_GSS INIT failed after recovery: {}  --  AUTH_SYS fallback", e);
                            self.gss_ctx = None;
                        }
                    }
                }
                Err(e) => {
                    warn!("NFS bypass: GSS re-establishment failed: {}  --  AUTH_SYS fallback", e);
                    self.gss_ctx = None;
                }
            }
        }

        info!("NFS bypass: session recovered (client_id={:#x})", self.client_id);
        Ok(())
    }

    /// Refresh the GSS context if it is expired or near expiry.
    /// No-op when the krb5 feature is disabled or no GSS context is active.
    pub fn ensure_gss_valid(&mut self) {
        #[cfg(feature = "krb5")]
        {
            let needs_refresh = self.gss_ctx.as_mut().map(|g| !g.is_valid()).unwrap_or(false);
            if needs_refresh {
                self.refresh_gss_context();
            }
        }
    }

    /// Unwrap GSS-protected reply data before parsing.
    /// No-op without krb5 feature or when no GSS context is active.
    fn unwrap_reply(&mut self, data: Vec<u8>) -> Result<Vec<u8>, NfsError> {
        #[cfg(feature = "krb5")]
        if let Some(ref mut gss) = self.gss_ctx {
            return rpc::unwrap_gss_reply(&data, gss);
        }
        Ok(data)
    }

    fn build_compound_msg(&mut self, xid: u32, tag: &str, ops: &[Nfs4Op]) -> Result<Vec<u8>, NfsError> {
        #[cfg(feature = "krb5")]
        {
            let needs_refresh = self.gss_ctx.as_mut().map(|g| !g.is_valid()).unwrap_or(false);
            if needs_refresh {
                self.refresh_gss_context();
            }
            if let Some(ref mut gss) = self.gss_ctx {
                return rpc::build_compound_gss(xid, tag, ops, gss);
            }
        }
        Ok(rpc::build_compound(xid, tag, self.uid, self.gid, &self.machine, ops))
    }

    /// Perform the RPCSEC_GSS INIT handshake to obtain the server-assigned
    /// context handle (RFC 2203 S5.2.2). Loops for multi-round GSS exchanges.
    #[cfg(feature = "krb5")]
    fn do_rpcsec_gss_init(&mut self, gss: &mut super::gss::GssContext) -> Result<(), NfsError> {
        const GSS_S_COMPLETE: u32 = 0;
        const GSS_S_CONTINUE_NEEDED: u32 = 1;

        let mut handle = Vec::<u8>::new();
        let mut token = gss.init_token.clone();

        loop {
            let xid = self.next_xid();
            let msg = rpc::build_rpcsec_gss_init(xid, &handle, &token);
            let reply = self.rpc_call(xid, &msg)?;
            let (server_handle, gss_major, gss_minor, _seq_window, cont_token) =
                rpc::parse_rpcsec_gss_init_reply(&reply)?;

            handle = server_handle;

            match gss_major {
                GSS_S_COMPLETE => {
                    // Server is done, but client may need one final
                    // gss_init_sec_context() to process the server's reply token.
                    if !cont_token.is_empty() {
                        let _ = gss.step(&cont_token);
                    }
                    gss.server_handle = handle;
                    debug!(
                        "RPCSEC_GSS INIT complete, handle={} bytes",
                        gss.server_handle.len()
                    );
                    return Ok(());
                }
                GSS_S_CONTINUE_NEEDED => {
                    match gss.step(&cont_token)? {
                        Some(new_token) => {
                            token = new_token;
                        }
                        None => {
                            return Err(NfsError::RpcError(
                                "GSS_S_CONTINUE_NEEDED but no output token from gss.step()"
                                    .into(),
                            ));
                        }
                    }
                }
                _ => {
                    return Err(NfsError::RpcError(format!(
                        "RPCSEC_GSS INIT failed: major={} minor={}",
                        gss_major, gss_minor
                    )));
                }
            }
        }
    }

    /// Re-establish an expired GSS context from system keytab/ccache.
    /// On failure, clears gss_ctx to fall back to AUTH_SYS.
    #[cfg(feature = "krb5")]
    fn refresh_gss_context(&mut self) {
        #[cfg(feature = "tls")]
        let channel_binding = self.stream.export_channel_binding();
        #[cfg(not(feature = "tls"))]
        let channel_binding: Option<Vec<u8>> = None;

        let hostname = self.server_info.server_hostname.clone();
        match super::gss::GssContext::establish(
            &hostname,
            self.server_info.security,
            channel_binding.as_deref(),
        ) {
            Ok(mut new_ctx) => {
                match self.do_rpcsec_gss_init(&mut new_ctx) {
                    Ok(()) => {
                        debug!("NFS bypass: GSS context refreshed (sec={})", self.server_info.security);
                        self.gss_ctx = Some(new_ctx);
                    }
                    Err(e) => {
                        warn!("NFS bypass: GSS context refresh INIT failed: {e}  --  falling back to AUTH_SYS");
                        self.gss_ctx = None;
                    }
                }
            }
            Err(e) => {
                warn!("NFS bypass: GSS context refresh failed: {e}  --  falling back to AUTH_SYS");
                self.gss_ctx = None;
            }
        }
    }

    /// Connect to an NFS server from a privileged source port (<1024).
    /// NFS servers with `secure` (default) reject connections from ports >=1024.
    fn connect_privileged(server: &std::net::SocketAddr) -> Result<TcpStream, NfsError> {
        use std::os::unix::io::FromRawFd;

        let domain = if server.is_ipv4() { libc::AF_INET } else { libc::AF_INET6 };
        // SAFETY: creating a TCP socket. No preconditions beyond valid domain.
        let fd = unsafe { libc::socket(domain, libc::SOCK_STREAM, 0) };
        if fd < 0 {
            return Err(NfsError::ConnectionFailed(std::io::Error::last_os_error()));
        }

        // Try binding to ports 900-1023 (privileged range, avoids well-known services)
        let mut bound = false;
        for port in (constants::NFS_PRIVILEGED_PORT_START..constants::NFS_PRIVILEGED_PORT_END).rev() {
            let ret = if server.is_ipv4() {
                let addr = libc::sockaddr_in {
                    sin_family: libc::AF_INET as u16,
                    sin_port: port.to_be(),
                    sin_addr: libc::in_addr { s_addr: 0 },
                    sin_zero: [0; 8],
                };
                // SAFETY: fd is a valid socket, addr is a properly initialized sockaddr_in.
                unsafe { libc::bind(fd, &addr as *const _ as *const libc::sockaddr, std::mem::size_of::<libc::sockaddr_in>() as u32) }
            } else {
                let addr = libc::sockaddr_in6 {
                    sin6_family: libc::AF_INET6 as u16,
                    sin6_port: port.to_be(),
                    sin6_flowinfo: 0,
                    sin6_addr: libc::in6_addr { s6_addr: [0; 16] },
                    sin6_scope_id: 0,
                };
                // SAFETY: fd is a valid socket, addr is a properly initialized sockaddr_in6.
                unsafe { libc::bind(fd, &addr as *const _ as *const libc::sockaddr, std::mem::size_of::<libc::sockaddr_in6>() as u32) }
            };
            if ret == 0 {
                bound = true;
                debug!("NFS bypass: bound to privileged port {}", port);
                break;
            }
        }

        if !bound {
            // SAFETY: fd is a valid socket descriptor.
            unsafe { libc::close(fd); }
            // Fall back to ephemeral port (works if server has `insecure` export option)
            debug!("NFS bypass: no privileged port available, using ephemeral");
            return Ok(TcpStream::connect_timeout(server, std::time::Duration::from_secs(constants::NFS_TCP_CONNECT_TIMEOUT_SECS))?);
        }

        // Connect to server
        let connect_result = match server {
            std::net::SocketAddr::V4(v4) => {
                let addr = libc::sockaddr_in {
                    sin_family: libc::AF_INET as u16,
                    sin_port: v4.port().to_be(),
                    sin_addr: libc::in_addr { s_addr: u32::from_ne_bytes(v4.ip().octets()) },
                    sin_zero: [0; 8],
                };
                // SAFETY: fd is a valid bound socket, addr is a properly initialized sockaddr_in.
                unsafe { libc::connect(fd, &addr as *const _ as *const libc::sockaddr, std::mem::size_of::<libc::sockaddr_in>() as u32) }
            }
            std::net::SocketAddr::V6(v6) => {
                let addr = libc::sockaddr_in6 {
                    sin6_family: libc::AF_INET6 as u16,
                    sin6_port: v6.port().to_be(),
                    sin6_flowinfo: v6.flowinfo(),
                    sin6_addr: libc::in6_addr { s6_addr: v6.ip().octets() },
                    sin6_scope_id: v6.scope_id(),
                };
                // SAFETY: fd is a valid bound socket, addr is a properly initialized sockaddr_in6.
                unsafe { libc::connect(fd, &addr as *const _ as *const libc::sockaddr, std::mem::size_of::<libc::sockaddr_in6>() as u32) }
            }
        };

        if connect_result != 0 {
            let err = std::io::Error::last_os_error();
            // SAFETY: fd is a valid socket descriptor.
            unsafe { libc::close(fd); }
            return Err(NfsError::ConnectionFailed(err));
        }

        // SAFETY: fd is a valid connected socket. Ownership transfers to TcpStream.
        Ok(unsafe { TcpStream::from_raw_fd(fd) })
    }

    /// Attempt RFC 9289 STARTTLS upgrade on the current TCP connection.
    ///
    /// Sends an AUTH_TLS NULL probe. If the server responds with "STARTTLS",
    /// performs a TLS 1.3 handshake via rustls with ALPN "sunrpc" and upgrades
    /// `self.stream` from `NfsTransport::Tcp` to `NfsTransport::Tls`.
    ///
    /// Returns `Ok(())` on successful upgrade, `Err(NfsError::TlsNotSupported)`
    /// if the server does not support RPC-over-TLS, or `Err(NfsError::TlsHandshakeFailed)`
    /// on TLS negotiation errors.
    #[cfg(feature = "tls")]
    fn try_tls_upgrade(&mut self) -> Result<(), NfsError> {
        use rustls::pki_types::ServerName;
        use rustls::{ClientConfig, ClientConnection, RootCertStore, StreamOwned};
        use std::sync::Arc;

        // Step 1: Send AUTH_TLS probe (RFC 9289 S4.1)
        let probe_xid = self.next_xid();
        let probe = super::rpc::build_auth_tls_probe(probe_xid);
        let reply = self.rpc_call(probe_xid, &probe)
            .map_err(|e| NfsError::TlsHandshakeFailed(format!("AUTH_TLS probe: {e}")))?;

        // Step 3: Check for STARTTLS in verifier body
        if !super::rpc::parse_auth_tls_reply(&reply) {
            return Err(NfsError::TlsNotSupported);
        }

        info!(
            "NFS TLS: server {} supports RPC-over-TLS, upgrading connection (RFC 9289)",
            self.server_info.server_addr
        );

        // Step 4: Build rustls ClientConfig  --  system CA store + ALPN "sunrpc"
        let roots = {
            let mut store = RootCertStore::empty();
            for cert in rustls_native_certs::load_native_certs().certs {
                let _ = store.add(cert);
            }
            store
        };
        let mut config = ClientConfig::builder()
            .with_root_certificates(roots)
            .with_no_client_auth();
        config.alpn_protocols = vec![constants::NFS_TLS_ALPN.to_vec()];
        #[cfg(feature = "ktls")]
        {
            config.enable_secret_extraction = true;
        }
        let config = Arc::new(config);

        // Step 5: Parse server hostname for certificate validation
        let hostname = &self.server_info.server_hostname;
        let server_name = ServerName::try_from(hostname.as_str())
            .map_err(|e| {
                NfsError::TlsHandshakeFailed(format!(
                    "invalid server hostname '{}' for TLS SNI: {e}",
                    hostname
                ))
            })?
            .to_owned();

        // Step 6: Clone the TCP socket fd for rustls ownership
        // try_clone() creates a dup(2) fd pointing to the same socket.
        // The original fd in self.stream is dropped when we reassign below.
        let tcp_clone = self
            .stream
            .tcp_ref()
            .try_clone()
            .map_err(|e| NfsError::TlsHandshakeFailed(format!("TCP clone: {e}")))?;

        // Step 7: Create rustls connection and wrap the TCP stream
        // The TLS handshake is driven lazily on the first write (EXCHANGE_ID compound).
        let conn = ClientConnection::new(config, server_name)
            .map_err(|e| NfsError::TlsHandshakeFailed(format!("rustls init: {e}")))?;
        let tls_stream = StreamOwned::new(conn, tcp_clone);

        // Step 8: Store the transport  --  with optional kTLS kernel offload.
        //
        // When the `ktls` feature is enabled we attempt to promote the
        // userspace TLS session into a kernel-offloaded (kTLS) socket.
        // `try_ktls_or_tls` consumes `tls_stream` on all code paths and
        // returns the appropriate `NfsTransport` variant:
        //   Ktls  --  promotion succeeded, kernel handles AES-GCM
        //   Tls   --  promotion failed pre-destructively, userspace TLS intact
        //   Tcp   --  promotion failed post-destructively (secrets consumed),
        //          degraded; next RPC error will trigger session recovery
        #[cfg(feature = "ktls")]
        {
            let (transport, promoted) = Self::try_ktls_or_tls(tls_stream);
            self.stream = transport;
            if promoted {
                info!(
                    "NFS TLS: kTLS kernel offload active  --  AES-GCM in-kernel, \
                     server={}",
                    self.server_info.server_addr
                );
                crate::metrics::NFS_BYPASS_KTLS_PROMOTED.inc();
            } else {
                info!(
                    "NFS TLS: TLS 1.3 handshake ready (will complete on first \
                     EXCHANGE_ID write), server={}, alpn=sunrpc",
                    self.server_info.server_addr
                );
            }
            return Ok(());
        }

        // Without ktls feature: always store as userspace TLS.
        #[cfg(not(feature = "ktls"))]
        {
            self.stream = NfsTransport::Tls(Box::new(tls_stream));
            info!(
                "NFS TLS: TLS 1.3 handshake ready (will complete on first EXCHANGE_ID write), \
                 server={}, alpn=sunrpc",
                self.server_info.server_addr
            );
        }
        Ok(())
    }

    /// Attempt kTLS kernel offload, falling back to userspace TLS.
    ///
    /// Consumes `tls_stream` on every path.  Returns `(transport, promoted)`
    /// where `promoted` is true only when kTLS was fully configured.
    #[cfg(feature = "ktls")]
    fn try_ktls_or_tls(
        mut tls_stream: rustls::StreamOwned<rustls::ClientConnection, TcpStream>,
    ) -> (NfsTransport, bool) {
        use rustls::StreamOwned;
        use std::os::unix::io::AsRawFd;

        // 1. Drive the TLS 1.3 handshake to completion (normally lazy).
        if let Err(e) = tls_stream.conn.complete_io(&mut tls_stream.sock) {
            debug!("NFS TLS: kTLS skipped  --  handshake completion failed ({e})");
            return (NfsTransport::Tls(Box::new(tls_stream)), false);
        }

        let fd = tls_stream.sock.as_raw_fd();

        // 2. Enable TLS ULP on the socket (non-destructive probe).
        //    Fails with ENOENT when CONFIG_TLS is absent.
        const SOL_TCP: libc::c_int = 6;
        const TCP_ULP_OPT: libc::c_int = 31;
        let ret = unsafe {
            libc::setsockopt(fd, SOL_TCP, TCP_ULP_OPT, b"tls\0".as_ptr().cast(), 4)
        };
        if ret < 0 {
            let err = std::io::Error::last_os_error();
            debug!("NFS TLS: kTLS ULP unavailable ({err}), using userspace TLS");
            return (NfsTransport::Tls(Box::new(tls_stream)), false);
        }

        // 3. Record cipher suite before consuming the connection.
        let suite = match tls_stream.conn.negotiated_cipher_suite() {
            Some(s) => s,
            None => {
                debug!("NFS TLS: kTLS skipped  --  no negotiated cipher suite");
                return (NfsTransport::Tls(Box::new(tls_stream)), false);
            }
        };

        // -- POINT OF NO RETURN --
        // `dangerous_extract_secrets()` consumes `ClientConnection`.
        // After this, we cannot fall back to userspace TLS.
        let StreamOwned { conn, sock: tcp } = tls_stream;

        let secrets = match conn.dangerous_extract_secrets() {
            Ok(s) => s,
            Err(e) => {
                debug!("NFS TLS: kTLS secret extraction failed ({e}), degraded to plain TCP");
                return (NfsTransport::Tcp(tcp), false);
            }
        };

        // 4. Build kernel crypto info from rustls secrets.
        let tx_info = match ktls::CryptoInfo::from_rustls(suite, secrets.tx) {
            Ok(i) => i,
            Err(e) => {
                debug!("NFS TLS: kTLS TX crypto unsupported ({e:?}), degraded to plain TCP");
                return (NfsTransport::Tcp(tcp), false);
            }
        };
        let rx_info = match ktls::CryptoInfo::from_rustls(suite, secrets.rx) {
            Ok(i) => i,
            Err(e) => {
                debug!("NFS TLS: kTLS RX crypto unsupported ({e:?}), degraded to plain TCP");
                return (NfsTransport::Tcp(tcp), false);
            }
        };

        // 5. Configure TX and RX on the socket fd.
        const SOL_TLS: libc::c_int = 282;
        const TLS_TX: libc::c_int = 1;
        const TLS_RX: libc::c_int = 2;

        let tx_ret = unsafe {
            libc::setsockopt(
                fd, SOL_TLS, TLS_TX,
                tx_info.as_ptr(), tx_info.size() as libc::socklen_t,
            )
        };
        if tx_ret < 0 {
            let err = std::io::Error::last_os_error();
            debug!("NFS TLS: kTLS TX setsockopt failed ({err}), degraded to plain TCP");
            return (NfsTransport::Tcp(tcp), false);
        }

        let rx_ret = unsafe {
            libc::setsockopt(
                fd, SOL_TLS, TLS_RX,
                rx_info.as_ptr(), rx_info.size() as libc::socklen_t,
            )
        };
        if rx_ret < 0 {
            let err = std::io::Error::last_os_error();
            debug!("NFS TLS: kTLS RX setsockopt failed ({err}), TX active but RX not  --  degraded");
            return (NfsTransport::Tcp(tcp), false);
        }

        (NfsTransport::Ktls(tcp), true)
    }

    fn next_xid(&self) -> u32 {
        self.xid_counter.fetch_add(1, Ordering::Relaxed)
    }

    fn next_sequence_id(&self) -> u32 {
        self.sequence_id.fetch_add(1, Ordering::Relaxed)
    }

    /// Send an RPC message body and receive the reply.
    ///
    /// `body` is the XDR-encoded RPC call **without** the 4-byte record mark.
    /// For TCP/TLS/kTLS transports this method prepends the record-mark frame
    /// (last-fragment bit + length).  The future `NfsTransport::Rdma` variant
    /// will use RFC 8166 `rdma_msg` framing instead.
    ///
    /// `_xid` is used by the RDMA transport (Wave 2) where the transaction ID
    /// appears in the RFC 8166 rdma_msg header separately from the XDR body.
    fn rpc_call(&mut self, _xid: u32, body: &[u8]) -> Result<Vec<u8>, NfsError> {
        // RDMA early-return: bypass TCP record-mark framing entirely.
        // RdmaConnection::rpc_call() builds the rdma_msg header and does
        // RDMA Send/Recv instead of stream write/read.
        #[cfg(feature = "rdma")]
        if let NfsTransport::Rdma(ref mut rdma) = self.stream {
            return rdma.rpc_call(_xid, body);
        }

        let frame = rpc::frame_rpc_message(body);
        self.stream.write_all(&frame)?;
        self.stream.flush()?;
        self.read_reply()
    }

    /// Write a single file to the NFS server via compound RPC.
    ///
    /// SEQUENCE + PUTFH(parent) + OPEN(create) + WRITE(FILE_SYNC) + CLOSE
    /// Retries on NFS4ERR_GRACE (server grace period after session creation).
    pub fn write_file(
        &mut self,
        parent_handle: &[u8],
        filename: &str,
        data: &[u8],
        mode: u32,
        uid: u32,
        gid: u32,
        mtime: (i64, i64),
    ) -> Result<(), NfsError> {
        for attempt in 0..constants::NFS_GRACE_RETRY_COUNT {
            match self.write_file_inner(parent_handle, filename, data, mode, uid, gid, mtime) {
                Ok(()) => return Ok(()),
                Err(NfsError::Nfs4Error { code: 10013, .. }) => {
                    // NFS4ERR_GRACE  --  server in grace period, retry after delay
                    if attempt < constants::NFS_GRACE_RETRY_COUNT - 1 {
                        debug!("NFS bypass: server in grace period, retry {} in {}ms", attempt + 1, constants::NFS_RECONNECT_SLEEP_MS);
                        std::thread::sleep(std::time::Duration::from_millis(constants::NFS_RECONNECT_SLEEP_MS));
                        continue;
                    }
                    return Err(NfsError::Nfs4Error {
                        code: 10013,
                        message: "server grace period persists after retries".into(),
                    });
                }
                Err(e) => return Err(e),
            }
        }
        Err(NfsError::Nfs4Error {
            code: 10013,
            message: "server grace period persists after retries".into(),
        })
    }

    fn write_file_inner(
        &mut self,
        parent_handle: &[u8],
        filename: &str,
        data: &[u8],
        mode: u32,
        _uid: u32,
        _gid: u32,
        _mtime: (i64, i64),
    ) -> Result<(), NfsError> {
        #[cfg(feature = "krb5")]
        if let Some(ref gss) = self.gss_ctx {
            if gss.rpc_gss_service() == super::gss::RpcGssService::Privacy
                && data.len() > constants::KRB5P_MAX_WRITE_DATA
            {
                return Err(NfsError::Nfs4Error {
                    code: rpc::NFS4ERR_INVAL,
                    message: format!(
                        "krb5p bypass: {}B exceeds {}B threshold (AES block boundary at priv_bytes=3952)",
                        data.len(), constants::KRB5P_MAX_WRITE_DATA
                    ),
                });
            }
        }

        let xid = self.next_xid();
        let seq_id = self.next_sequence_id();

        let ops = vec![
            Nfs4Op::Sequence {
                session_id: self.session_id,
                sequence_id: seq_id,
                slot_id: 0,
                highest_slot_id: 0,
                cache_this: false,
            },
            Nfs4Op::PutFh { handle: parent_handle.to_vec() },
            Nfs4Op::Open {
                seqid: 0,
                share_access: rpc::OPEN4_SHARE_ACCESS_WRITE,
                share_deny: rpc::OPEN4_SHARE_DENY_NONE,
                clientid: self.client_id,
                owner: format!("foxing-{}", std::process::id()).into_bytes(),
                filename: filename.to_string(),
                mode,
                create_owner: None,
                create_group: None,
                create_mtime: None,
            },
            Nfs4Op::Write {
                stateid: StateId::current(),
                offset: 0,
                stable: WriteStable::FileSync,
                data: data.to_vec(),
            },
            Nfs4Op::Close {
                seqid: 1,
                stateid: StateId::current(),
            },
        ];

        let msg = self.build_compound_msg(xid, "fxcp", &ops)?;

        debug!("NFS write compound: {} bytes, {} ops, handle={} bytes, data={} bytes",
               msg.len(), ops.len(), parent_handle.len(), data.len());
        if msg.len() > 140 {
            debug!("NFS compound hex (first 200 bytes): {:02x?}", &msg[..msg.len().min(200)]);
        }

        let reply_data = self.rpc_call(xid, &msg)?;
        let reply_data = self.unwrap_reply(reply_data)?;
        let reply = rpc::parse_compound_reply(&reply_data)?;

        // Log detailed reply info
        debug!("NFS compound reply: overall_status={} ({}) ops={}",
               reply.status, rpc::nfs4_error_name(reply.status), reply.op_results.len());
        for (i, r) in reply.op_results.iter().enumerate() {
            debug!("  op[{}]: op={} status={} ({})", i, r.op, r.status, rpc::nfs4_error_name(r.status));
        }

        if reply.status != rpc::NFS4_OK {
            let code = reply.status;
            // Check for recoverable errors
            if code == rpc::NFS4ERR_BADSESSION || code == rpc::NFS4ERR_BADSEQ {
                return Err(NfsError::Nfs4Error {
                    code,
                    message: format!("session error: {}  --  needs recovery", rpc::nfs4_error_name(code)),
                });
            }
            if code == rpc::NFS4ERR_STALE {
                return Err(NfsError::StaleHandle {
                    path: filename.to_string(),
                });
            }
            return Err(NfsError::Nfs4Error {
                code,
                message: format!("compound failed: {}", rpc::nfs4_error_name(code)),
            });
        }

        // Check individual op statuses
        for result in &reply.op_results {
            if result.status != rpc::NFS4_OK {
                return Err(NfsError::Nfs4Error {
                    code: result.status,
                    message: format!("op {} failed: {}", result.op, rpc::nfs4_error_name(result.status)),
                });
            }
        }

        debug!("NFS bypass: wrote {} ({} bytes) via compound RPC", filename, data.len());
        Ok(())
    }

    /// Write multiple files to the NFS server in a single compound RPC.
    ///
    /// Builds: SEQUENCE + PUTFH(parent) + [OPEN+WRITE+CLOSE]xN
    /// Returns `Vec<Result<(), NfsError>>`  --  per-file results for partial failure handling.
    ///
    /// Max 4 files per compound (SEQUENCE + PUTFH + 4x(OPEN+WRITE+CLOSE) = 14 ops,
    /// within `ca_maxoperations=16` negotiated in CREATE_SESSION). Larger batches are
    /// automatically split into multiple compounds.
    pub fn write_files_compound(
        &mut self,
        parent_handle: &[u8],
        files: &[(&str, &[u8], u32, u32, u32, (i64, i64))],  // (filename, data, mode, uid, gid, mtime)
    ) -> Result<Vec<Result<(), NfsError>>, NfsError> {
        if files.is_empty() { return Ok(vec![]); }

        // SEQUENCE(1) + PUTFH(1) + N*(OPEN+WRITE+CLOSE=3) = 2 + 3N
        // AIMD tuner grows from 3 toward (ca_maxops-2)/3 after sustained success.
        let max_files = self.compound_tuner.current();
        // Byte-size gate: under krb5p privacy, compound_args + 4 (seq_num) must stay
        // under 3,952 bytes (the AES block boundary where kernel gss_unwrap breaks).
        // Under AUTH_SYS/krb5/krb5i, use ~800KB (80% of typical svc_max_payload).
        #[cfg(feature = "krb5")]
        let compound_byte_limit: usize = if self.gss_ctx.as_ref()
            .map(|g| g.rpc_gss_service() == super::gss::RpcGssService::Privacy)
            .unwrap_or(false)
        {
            constants::KRB5P_MAX_WRITE_DATA + constants::GSS_WRAP_OVERHEAD_BYTES
        } else {
            constants::NFS_COMPOUND_BYTE_LIMIT
        };
        #[cfg(not(feature = "krb5"))]
        let compound_byte_limit: usize = constants::NFS_COMPOUND_BYTE_LIMIT;

        let effective_max = {
            let mut byte_acc: usize = constants::NFS_XDR_HEADER_OVERHEAD;
            let mut count = 0usize;
            for (_, data, ..) in files.iter() {
                byte_acc += data.len() + constants::NFS_XDR_PER_OP_OVERHEAD;
                if byte_acc > compound_byte_limit {
                    break;
                }
                count += 1;
            }
            count.max(1).min(max_files)
        };
        if files.len() > effective_max {
            let mut all_results = Vec::with_capacity(files.len());
            for chunk in files.chunks(effective_max) {
                let results = self.write_files_compound(parent_handle, chunk)?;
                all_results.extend(results);
            }
            return Ok(all_results);
        }

        let xid = self.next_xid();
        let seq_id = self.next_sequence_id();

        let mut ops = vec![
            Nfs4Op::Sequence {
                session_id: self.session_id,
                sequence_id: seq_id,
                slot_id: 0,
                highest_slot_id: 0,
                cache_this: false,
            },
            Nfs4Op::PutFh { handle: parent_handle.to_vec() },
        ];

        let owner = format!("foxing-{}", std::process::id()).into_bytes();
        for (filename, data, mode, _uid, _gid, _mtime) in files {
            ops.push(Nfs4Op::Open {
                seqid: 0,
                share_access: rpc::OPEN4_SHARE_ACCESS_WRITE,
                share_deny: rpc::OPEN4_SHARE_DENY_NONE,
                clientid: self.client_id,
                owner: owner.clone(),
                filename: filename.to_string(),
                mode: *mode,
                create_owner: None,
                create_group: None,
                create_mtime: None,
            });
            ops.push(Nfs4Op::Write {
                stateid: StateId::current(),
                offset: 0,
                stable: WriteStable::FileSync,
                data: data.to_vec(),
            });
            ops.push(Nfs4Op::Close {
                seqid: 1,
                stateid: StateId::current(),
            });
        }

        let msg = self.build_compound_msg(xid, "mwrt", &ops)?;
        let reply_data = self.rpc_call(xid, &msg)?;
        let reply_data = self.unwrap_reply(reply_data)?;
        let reply = rpc::parse_compound_reply(&reply_data)?;

        // Parse per-file results from compound reply.
        // Layout: [0]=SEQUENCE, [1]=PUTFH, then groups of 3 ops per file:
        //   [2+i*3]=OPEN, [3+i*3]=WRITE, [4+i*3]=CLOSE
        let mut results = Vec::with_capacity(files.len());
        for file_idx in 0..files.len() {
            let open_idx = 2 + file_idx * 3;
            let mut file_ok = true;
            for op_offset in 0..3 {
                let idx = open_idx + op_offset;
                match reply.op_results.get(idx) {
                    Some(r) if r.status == rpc::NFS4_OK => {}
                    Some(r) => {
                        file_ok = false;
                        // NFS4ERR_REQ_TOO_BIG (10019) / NFS4ERR_REP_TOO_BIG (10028)
                        if r.status == 10019 || r.status == 10028 {
                            self.compound_tuner.on_size_error();
                        }
                        results.push(Err(NfsError::Nfs4Error {
                            code: r.status,
                            message: format!(
                                "multi-write file {} op {} failed: {}",
                                file_idx, r.op, rpc::nfs4_error_name(r.status)
                            ),
                        }));
                        break;
                    }
                    None => {
                        file_ok = false;
                        results.push(Err(NfsError::Nfs4Error {
                            code: reply.status,
                            message: format!(
                                "multi-write file {} truncated compound (prior error)",
                                file_idx
                            ),
                        }));
                        break;
                    }
                }
            }
            if file_ok {
                debug!("NFS multi-write: file {} ({}) ok", file_idx, files[file_idx].0);
                results.push(Ok(()));
            }
        }
        if results.iter().all(|r| r.is_ok()) {
            self.compound_tuner.on_success();
        }
        Ok(results)
    }

    /// Read a complete RPC reply (record-mark framed) from the TCP stream.
    /// Handles multi-fragment replies by reading until the last-fragment bit is set.
    fn read_reply(&mut self) -> Result<Vec<u8>, NfsError> {
        set_quickack(&self.stream);
        let mut result = Vec::with_capacity(constants::NFS_REPLY_INITIAL_CAPACITY);

        loop {
            let mut rm_buf = [0u8; 4];
            self.stream.read_exact(&mut rm_buf)?;
            let rm = u32::from_be_bytes(rm_buf);
            let last_fragment = (rm & 0x80000000) != 0;
            let length = (rm & 0x7FFFFFFF) as usize;

            if length > crate::constants::NFS_MAX_REQUEST_SIZE {
                return Err(NfsError::RpcError(format!("reply fragment too large: {} bytes", length)));
            }

            if result.is_empty() {
                // First fragment  --  include the record mark for the parser
                result.extend_from_slice(&rm_buf);
            }

            let offset = result.len();
            result.resize(offset + length, 0);
            self.stream.read_exact(&mut result[offset..])?;

            if last_fragment {
                break;
            }
        }

        Ok(result)
    }

    /// Get or resolve the NFS file handle for a directory path.
    ///
    /// Uses PUTROOTFH + LOOKUP compounds to walk from the export root to
    /// the target directory, caching intermediate handles.
    pub fn get_or_resolve_handle(&mut self, dir_path: &Path) -> Result<Vec<u8>, NfsError> {
        if let Some(cached) = self.dir_handle_cache.get(dir_path) {
            return Ok(cached.clone());
        }

        // Compute relative path from mount point
        let rel = dir_path.strip_prefix(&self.server_info.mount_point)
            .unwrap_or(std::path::Path::new(""));

        // Walk from export root via PUTROOTFH + LOOKUP chain + GETFH
        let handle = self.resolve_via_lookup(rel)?;

        self.dir_handle_cache.insert(dir_path.to_path_buf(), handle.clone());
        Ok(handle)
    }

    /// Resolve a relative path from the export root using PUTFH + LOOKUP + GETFH.
    fn resolve_via_lookup(&mut self, rel_path: &Path) -> Result<Vec<u8>, NfsError> {
        let xid = self.next_xid();
        let seq_id = self.next_sequence_id();

        // Get the mount point's NFS wire filehandle via name_to_handle_at.
        // The kernel stores the server's opaque handle; we extract the wire
        // portion (skipping the kernel's internal 14-byte header).
        let mount_fh = super::mount::resolve_nfs_handle(&self.server_info.mount_point)
            .map_err(|e| NfsError::StaleHandle { path: format!("mount: {}", e) })?;
        debug!("NFS PUTFH: mount handle {} bytes: {:02x?}", mount_fh.len(), &mount_fh[..mount_fh.len().min(16)]);

        let mut ops = vec![
            Nfs4Op::Sequence {
                session_id: self.session_id,
                sequence_id: seq_id,
                slot_id: 0,
                highest_slot_id: 0,
                cache_this: false,
            },
            Nfs4Op::PutFh { handle: mount_fh },
        ];

        // LOOKUP only the relative path within the export
        for component in rel_path.components() {
            if let std::path::Component::Normal(name) = component {
                ops.push(Nfs4Op::Lookup { name: name.to_string_lossy().into_owned() });
            }
        }

        // GETFH to retrieve the actual server-side filehandle
        ops.push(Nfs4Op::GetFh);

        let msg = self.build_compound_msg(xid, "lkup", &ops)?;
        let reply_data = self.rpc_call(xid, &msg)?;
        let reply_data = self.unwrap_reply(reply_data)?;
        let reply = rpc::parse_compound_reply(&reply_data)?;

        debug!("NFS LOOKUP reply: overall_status={} ({}) ops={}",
               reply.status, rpc::nfs4_error_name(reply.status), reply.op_results.len());
        for (i, r) in reply.op_results.iter().enumerate() {
            debug!("  LOOKUP op[{}]: op={} status={} ({})", i, r.op, r.status, rpc::nfs4_error_name(r.status));
        }

        if reply.status != rpc::NFS4_OK {
            return Err(NfsError::Nfs4Error {
                code: reply.status,
                message: format!("LOOKUP for {:?} failed: {}", rel_path, rpc::nfs4_error_name(reply.status)),
            });
        }

        // Extract the filehandle from GETFH result
        for result in &reply.op_results {
            if result.op == rpc::OP_GETFH
                && let Some(fh) = &result.filehandle {
                    debug!("NFS GETFH: resolved {:?} -> {} bytes", rel_path, fh.len());
                    return Ok(fh.clone());
                }
        }

        Err(NfsError::SessionFailed(format!("GETFH not found in reply for {:?}", rel_path)))
    }

    /// Invalidate a cached directory handle.
    pub fn invalidate_handle(&self, dir_path: &Path) {
        self.dir_handle_cache.remove(dir_path);
    }

    /// Batch-fetch size+mtime for multiple files in a directory.
    ///
    /// Sends a single compound: SEQUENCE + PUTFH(dir) + [LOOKUP(file) + GETATTR]xN
    /// Returns (filename, size, mtime_sec, mtime_nsec) for each file found.
    /// Files that don't exist (LOOKUP returns NFS4ERR_NOENT) are silently skipped.
    /// Max 7 files per compound (SEQUENCE + PUTFH + 7x(LOOKUP+GETATTR) = 16 ops).
    pub fn batch_stat(
        &mut self,
        dir_handle: &[u8],
        filenames: &[&str],
    ) -> Result<Vec<(String, u64, i64, i64)>, NfsError> {
        let xid = self.next_xid();
        let seq_id = self.next_sequence_id();

        let attr_request = [1 << rpc::FATTR4_SIZE, 1 << (rpc::FATTR4_TIME_MODIFY - 32)];

        let mut ops = vec![
            rpc::Nfs4Op::Sequence {
                session_id: self.session_id,
                sequence_id: seq_id,
                slot_id: 0,
                highest_slot_id: 0,
                cache_this: false,
            },
            rpc::Nfs4Op::PutFh { handle: dir_handle.to_vec() },
        ];

        for name in filenames {
            ops.push(rpc::Nfs4Op::Lookup { name: name.to_string() });
            ops.push(rpc::Nfs4Op::GetAttr { attr_request });
        }

        let msg = self.build_compound_msg(xid, "bstat", &ops)?;
        let reply_data = self.rpc_call(xid, &msg)?;
        let reply_data = self.unwrap_reply(reply_data)?;
        let reply = rpc::parse_compound_reply(&reply_data)?;

        // Extract results: for each LOOKUP+GETATTR pair, check status
        let mut results = Vec::new();
        for (i, name) in filenames.iter().enumerate() {
            // Find the LOOKUP result for this file (skip SEQUENCE + PUTFH = first 2 ops)
            let lookup_idx = 2 + i * 2;
            let getattr_idx = lookup_idx + 1;

            // Check LOOKUP status
            if let Some(lookup_result) = reply.op_results.get(lookup_idx) {
                if lookup_result.status != rpc::NFS4_OK {
                    continue; // file doesn't exist or LOOKUP failed
                }
            } else {
                break; // compound stopped processing (prior error)
            }

            // Check GETATTR result
            if let Some(getattr_result) = reply.op_results.get(getattr_idx) {
                if getattr_result.status == rpc::NFS4_OK
                    && let (Some(size), Some((mtime_s, mtime_ns))) = (getattr_result.size, getattr_result.mtime) {
                        results.push((name.to_string(), size, mtime_s, mtime_ns));
                    }
            } else {
                break;
            }
        }

        Ok(results)
    }

    /// Batch-read the first `read_bytes` of multiple files in a single compound RPC.
    ///
    /// Compound structure: SEQUENCE + PUTFH(dir) + [LOOKUP(name) + READ(offset=0, count=read_bytes)] × N
    /// Uses the anonymous stateid (seqid=0, other=all-zeros) — no OPEN/CLOSE needed.
    /// Files where LOOKUP returns NFS4ERR_NOENT are silently skipped.
    ///
    /// Returns: Vec of (filename, head_bytes) for files that were successfully read.
    pub fn batch_read_head(
        &mut self,
        dir_handle: &[u8],
        filenames: &[&str],
        read_bytes: usize,
    ) -> Result<Vec<(String, Vec<u8>)>, NfsError> {
        if filenames.is_empty() {
            return Ok(Vec::new());
        }

        let xid = self.next_xid();
        let seq_id = self.next_sequence_id();

        let mut ops = Vec::with_capacity(2 + filenames.len() * 2);
        ops.push(rpc::Nfs4Op::Sequence {
            session_id: self.session_id,
            sequence_id: seq_id,
            slot_id: 0,
            highest_slot_id: 0,
            cache_this: false,
        });
        ops.push(rpc::Nfs4Op::PutFh { handle: dir_handle.to_vec() });

        for name in filenames {
            ops.push(rpc::Nfs4Op::Lookup { name: name.to_string() });
            ops.push(rpc::Nfs4Op::Read {
                stateid: rpc::StateId::default(),
                offset: 0,
                count: read_bytes as u32,
            });
        }

        let msg = self.build_compound_msg(xid, "brdh", &ops)?;
        let reply_data = self.rpc_call(xid, &msg)?;
        let reply_data = self.unwrap_reply(reply_data)?;
        let reply = rpc::parse_compound_reply(&reply_data)?;

        let mut out = Vec::new();
        for (i, name) in filenames.iter().enumerate() {
            let lookup_idx = 2 + i * 2;
            let read_idx = lookup_idx + 1;

            // Check LOOKUP status
            if let Some(lookup_result) = reply.op_results.get(lookup_idx) {
                if lookup_result.status != rpc::NFS4_OK {
                    continue; // file not found or error — skip silently
                }
            } else {
                break; // compound stopped processing (prior error)
            }

            // Check READ result
            if let Some(read_result) = reply.op_results.get(read_idx) {
                if read_result.status == rpc::NFS4_OK
                    && let Some(ref data) = read_result.data {
                        out.push((name.to_string(), data.clone()));
                    }
            } else {
                break;
            }
        }

        Ok(out)
    }
}

/// Pool of NFS compound RPC sessions for parallel writes.
///
/// Each session has its own TCP connection and NFSv4.1 session ID,
/// enabling true parallel RPCs to the server (one in-flight per session).
/// Rayon threads grab sessions by index (round-robin) to avoid contention.
pub struct NfsClientPool {
    clients: Vec<parking_lot::Mutex<NfsCompoundClient>>,
}

impl NfsClientPool {
    /// Create a pool of `pool_size` independent NFS sessions.
    pub fn new(info: &NfsBypassInfo, pool_size: usize) -> Result<Self, NfsError> {
        let pool_size = pool_size.max(1);
        let mut clients = Vec::with_capacity(pool_size);
        for i in 0..pool_size {
            match NfsCompoundClient::connect(info) {
                Ok(client) => {
                    clients.push(parking_lot::Mutex::new(client));
                    #[cfg(feature = "tls")]
                    {
                        use super::NfsTransportSecurity;
                        if info.transport_security != NfsTransportSecurity::None {
                            info!("NFS pool: session {} TLS active", i);
                        }
                    }
                }
                Err(e) => {
                    if i == 0 {
                        return Err(e); // First session must succeed
                    }
                    info!("NFS pool: session {} failed ({}), pool size = {}", i, e, clients.len());
                    break; // Use whatever we got
                }
            }
        }
        info!("NFS pool: {} sessions established to {}", clients.len(), info.server_addr);
        Ok(Self { clients })
    }

    /// Get a session by index (round-robin across pool).
    pub fn get(&self, idx: usize) -> &parking_lot::Mutex<NfsCompoundClient> {
        &self.clients[idx % self.clients.len()]
    }

    /// Pool size (number of active sessions).
    pub fn len(&self) -> usize {
        self.clients.len()
    }

    /// Returns the [`NfsBypassInfo`] from the first session in the pool.
    pub fn bypass_info(&self) -> NfsBypassInfo {
        self.clients
            .first()
            .map(|s| s.lock().server_info.clone())
            .unwrap_or_default()
    }
}

/// Fast NFS server liveness check via NULL RPC.
///
/// Opens a fresh TCP connection to the NFS server and sends a NULL procedure
/// call. Returns true if the server responds within 2 seconds. Does not
/// touch any session or cached state  --  safe to call from the health probe.
pub fn probe_server_alive(server_addr: &std::net::SocketAddr) -> bool {
    use std::io::{Read, Write};
    let mut stream = match std::net::TcpStream::connect_timeout(server_addr, std::time::Duration::from_secs(constants::NFS_PROBE_TIMEOUT_SECS)) {
        Ok(s) => s,
        Err(_) => return false,
    };
    let _ = stream.set_read_timeout(Some(std::time::Duration::from_secs(constants::NFS_PROBE_TIMEOUT_SECS)));
    let _ = stream.set_write_timeout(Some(std::time::Duration::from_secs(constants::NFS_PROBE_TIMEOUT_SECS)));
    let msg = rpc::frame_rpc_message(&rpc::build_null_call(1));
    if stream.write_all(&msg).is_err() { return false; }
    if stream.flush().is_err() { return false; }
    // NULL reply is just an RPC reply header  --  reading any bytes means success
    let mut rm_buf = [0u8; 4];
    stream.read_exact(&mut rm_buf).is_ok()
}

/// Encode AUTH_SYS credentials.
fn encode_auth_sys(enc: &mut super::xdr::XdrEncoder, uid: u32, gid: u32, machine: &str) {
    enc.encode_u32(rpc::AUTH_SYS);
    let mut body = super::xdr::XdrEncoder::new(64);
    body.encode_u32(0);          // stamp
    body.encode_string(machine);
    body.encode_u32(uid);
    body.encode_u32(gid);
    body.encode_u32(1);          // gids count
    body.encode_u32(gid);        // gids[0]
    let body_bytes = body.into_bytes();
    enc.encode_opaque(&body_bytes);
}

#[cfg(test)]
mod tests {
    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
    #[cfg(feature = "tls")]
    #[test]
    fn tls_alpn_is_sunrpc() {
        // RFC 9289 S7.2: ALPN protocol identifier MUST be "sunrpc"
        assert_eq!(crate::constants::NFS_TLS_ALPN, b"sunrpc");
        assert_eq!(crate::constants::NFS_TLS_ALPN.len(), 6);
    }
}