nfs-rs 0.8.4

use bytes::{Buf, Bytes};
use tracing::{debug, warn};

use super::callback::RecallNotification;
use super::compound::OpenArgs;
use super::mount::{Mount41, decode_fh, extract_open_delegation, extract_stateid};
use super::pnfs_io::PnfsWriteOutcome;
use super::state::{AccessMode, StateId};
use crate::error::{NfsError, Result};
use crate::mount;
use crate::mount::{WriteOutcome, WriteStability};
use crate::nfs4::attrs::{decode_getattr_response, encode_setattr, standard_getattr_bitmap};

impl Mount41 {
    /// Try an UNSTABLE write through pNFS data servers. `None` means no DS
    /// mutation was attempted and the caller must write through the MDS.
    pub(crate) async fn write_pnfs(
        &self,
        fh: &Bytes,
        offset: u64,
        data: Bytes,
    ) -> Option<Result<mount::WriteOutcome>> {
        if let Err(error) = self.refresh_layout_for_write(fh, offset).await {
            return Some(Err(error));
        }
        let _io_guard = self.layout_manager.read_file_io(fh).await;
        // Recall/refresh commits hold the exclusive I/O gate. Inspect failure
        // only after settlement, while the layout is protected from replacement.
        if self
            .layout_manager
            .pending_writes(fh)
            .await
            .iter()
            .any(|w| w.failed() && !w.is_done())
        {
            return Some(Err(crate::NfsError::Rpc(
                "previous pNFS write failed; close and recover before writing again".into(),
            )));
        }
        match self.pnfs_write(fh, offset, data).await {
            PnfsWriteOutcome::NotAttempted => None,
            PnfsWriteOutcome::Attempted(result) => Some(result),
        }
    }

    /// WRITE to the MDS with the requested stability level; no COMMIT is
    /// issued here. pNFS layouts are bypassed so that one COMMIT on the MDS
    /// covers every chunk written this way.
    pub(crate) async fn write_how(
        &self,
        fh: Bytes,
        offset: u64,
        data: Bytes,
        stability: WriteStability,
    ) -> Result<WriteOutcome> {
        let _io_guard = self.layout_manager.read_file_io(&fh).await;
        let sid = self
            .state
            .has_open(&fh, AccessMode::Write)
            .await
            .unwrap_or_else(StateId::anonymous);
        let stateid = sid.raw;
        let data_len = data.len() as u32;
        let fh_ref = fh.clone();
        let resp = self
            .compound_write("write", data, |b| {
                b.require_generation(sid.generation)
                    .putfh(&fh_ref)
                    .write_header(&stateid, offset, stability.stable_how(), data_len)
            })
            .await?;
        resp.op_ok(1)?; // PUTFH
        let write_op = resp.op_ok(2)?;
        let mut d = write_op.data.clone();
        if d.remaining() < 16 {
            return Err(NfsError::Xdr("WRITE result too short".to_string()));
        }
        let count = d.get_u32();
        let committed = d.get_u32();
        let mut verifier = [0u8; 8];
        d.copy_to_slice(&mut verifier);
        Ok(WriteOutcome {
            pnfs: None,
            count,
            committed: crate::WriteCommitted::try_from(committed)?,
            verifier: Some(verifier),
        })
    }

    pub(crate) async fn open(
        &self,
        dir_fh: Bytes,
        filename: &str,
        access: u32,
    ) -> Result<mount::ObjRes> {
        let access_mode = match access {
            crate::OPEN_READ => AccessMode::Read,
            crate::OPEN_WRITE => AccessMode::Write,
            _ => AccessMode::Both,
        };

        let bitmap = standard_getattr_bitmap();
        let open_args = OpenArgs {
            seqid: 0,
            share_access: access_mode.share_access(),
            share_deny: 0, // OPEN4_SHARE_DENY_NONE
            client_id: self.session_holder.get().await.client_id(),
            owner: Bytes::from_static(b"nfs-rs"),
            create: false,
            create_attrs_mask: vec![],
            create_attrs_vals: vec![],
            claim_file: filename.to_string(),
            want_no_delegation: !self.retain_delegations,
        };
        let resp = self
            .compound("open", |b| {
                b.putfh(&dir_fh).open(&open_args).getfh().getattr(&bitmap)
            })
            .await?;
        resp.op_ok(1)?; // PUTFH
        let open_op = resp.op_ok(2)?; // OPEN
        let mut open_data = open_op.data.clone();
        let stateid = extract_stateid(&mut open_data)?;
        let delegation = extract_open_delegation(&mut open_data);

        let getfh = resp.op_ok(3)?; // GETFH
        let mut fh_data = getfh.data.clone();
        let fh = decode_fh(&mut fh_data)?;
        let getattr = resp.op_ok(4)?; // GETATTR
        let mut attr_data = getattr.data.clone();
        let attr = decode_getattr_response(&mut attr_data)?;

        if !self.retain_delegations {
            self.return_unsolicited_delegation(delegation, &fh);
        }

        // Register in StateManager for READ/WRITE to use
        self.state
            .register_open(
                &fh,
                StateId::from_bytes_at(&stateid, resp.session_generation),
                access_mode,
            )
            .await?;

        Ok(mount::ObjRes {
            fh,
            attr: Some(attr),
        })
    }

    /// server 无视 WANT_NO_DELEG 仍授予 delegation 时，立即入队异步
    /// DELEGRETURN（复用 recall handler 路径）。不等 server 后续
    /// CB_RECALL——多 client（如多 worker）并发访问同一文件时，recall
    /// 期间 server 会阻塞冲突请求，主动归还消除该阻塞窗口。
    fn return_unsolicited_delegation(&self, delegation: Option<[u8; 16]>, fh: &Bytes) {
        if let Some(deleg_sid) = delegation {
            let notification = RecallNotification::Delegation {
                stateid: deleg_sid,
                truncate: false,
                fh: fh.clone(),
            };
            // best-effort：队列满时放弃，server 稍后 CB_RECALL 仍能兜底归还
            if let Err(e) = self.recall_tx.try_send(notification) {
                debug!(error = %e, "proactive DELEGRETURN enqueue failed, deferring to CB_RECALL");
            }
        }
    }

    pub(crate) async fn open_path(&self, path: &str, access: u32) -> Result<mount::ObjRes> {
        let (dir, name) = crate::split_path(path)?;
        let dir_obj = self.lookup_path(&dir).await?;
        self.open(dir_obj.fh, &name, access).await
    }

    pub(crate) async fn close_file(&self, fh: Bytes) -> Result<()> {
        let _io_guard = self.layout_manager.write_file_io(&fh).await;
        // CLOSE 前提交累积的 layout 写入范围（size/mtime 对 MDS 可见）
        self.flush_layoutcommit(&fh).await?;
        // Return pNFS layout before CLOSE if server requested return_on_close
        if let Some(layout) = self.layout_manager.get_layout(&fh).await
            && layout.return_on_close
        {
            self.layoutreturn_file(&fh).await?;
        }
        // Release ref in StateManager; if ref_count hits 0, send CLOSE to server
        if let Some(sid) = self.state.release(&fh).await {
            self.compound("close", |b| {
                b.require_generation(sid.generation)
                    .putfh(&fh)
                    .close(0, &sid.raw)
            })
            .await?;
            self.state.close_succeeded(&fh, &sid).await;
        }
        Ok(())
    }

    pub(crate) async fn create(
        &self,
        dir_fh: Bytes,
        filename: &str,
        mode: Option<u32>,
    ) -> Result<mount::ObjRes> {
        self.create_with_access(dir_fh, filename, mode, crate::OPEN_BOTH)
            .await
    }

    pub(crate) async fn create_with_access(
        &self,
        dir_fh: Bytes,
        filename: &str,
        mode: Option<u32>,
        access: u32,
    ) -> Result<mount::ObjRes> {
        let access_mode = match access {
            crate::OPEN_READ => AccessMode::Read,
            crate::OPEN_WRITE => AccessMode::Write,
            crate::OPEN_BOTH => AccessMode::Both,
            _ => {
                return Err(NfsError::InvalidInput(format!(
                    "invalid OPEN access {access}"
                )));
            }
        };
        // 不在 OPEN createattrs 传 mode，避免 NFS4ERR_ATTRNOTSUPP；创建后 SETATTR 设置。
        let bitmap = standard_getattr_bitmap();
        let open_args = OpenArgs {
            seqid: 0,
            share_access: access_mode.share_access(),
            share_deny: 0, // OPEN4_SHARE_DENY_NONE
            client_id: self.session_holder.get().await.client_id(),
            owner: Bytes::from_static(b"nfs-rs-create"),
            create: true,
            create_attrs_mask: vec![],
            create_attrs_vals: vec![],
            claim_file: filename.to_string(),
            want_no_delegation: !self.retain_delegations,
        };
        let resp = self
            .compound("create", |b| {
                b.putfh(&dir_fh).open(&open_args).getfh().getattr(&bitmap)
            })
            .await?;
        resp.op_ok(1)?; // PUTFH
        let open_op = resp.op_ok(2)?; // OPEN
        // Extract stateid from OPEN result for CLOSE
        let mut open_data = open_op.data.clone();
        let stateid = extract_stateid(&mut open_data)?;
        let delegation = extract_open_delegation(&mut open_data);
        let getfh = resp.op_ok(3)?;
        let mut fh_data = getfh.data.clone();
        let fh = decode_fh(&mut fh_data)?;
        if !self.retain_delegations {
            self.return_unsolicited_delegation(delegation, &fh);
        }
        let getattr = resp.op_ok(4)?;
        let mut attr_data = getattr.data.clone();
        let mut attr = decode_getattr_response(&mut attr_data)?;
        // SETATTR with open stateid (before CLOSE) — RFC 5661 §18.30.4:
        // 使用 open stateid 避免与 delegation 冲突，且保证原子性。
        if let Some(m) = mode {
            let (attrmask, attr_vals) = encode_setattr(Some(m), None, None, None, None, None);
            match self
                .compound("setattr", |b| {
                    b.putfh(&fh).setattr(&stateid, &attrmask, &attr_vals)
                })
                .await
            {
                Err(e) => {
                    warn!(error = %e, mode = m, "create: SETATTR mode failed after file creation");
                }
                _ => {
                    attr.file_mode = m;
                }
            }
        }
        // 保持文件 open 并注册 stateid，由调用方 close_file() 时发 CLOSE 释放
        // （umount 时 drain 兜底）。后续 WRITE 因此持有真实 open stateid——
        // RFC 8881 §13.9.1 禁止 DS I/O 使用 special stateid，提前 CLOSE 会导致
        // pNFS 写全部 NFS4ERR_BAD_STATEID 回退 MDS。
        self.state
            .register_open(
                &fh,
                StateId::from_bytes_at(&stateid, resp.session_generation),
                access_mode,
            )
            .await?;
        Ok(mount::ObjRes {
            fh,
            attr: Some(attr),
        })
    }

    pub(crate) async fn create_path(&self, path: &str, mode: Option<u32>) -> Result<mount::ObjRes> {
        self.create_path_with_access(path, mode, crate::OPEN_BOTH)
            .await
    }

    pub(crate) async fn create_path_with_access(
        &self,
        path: &str,
        mode: Option<u32>,
        access: u32,
    ) -> Result<mount::ObjRes> {
        let (dir, name) = crate::split_path(path)?;
        let dir_obj = self.lookup_path(&dir).await?;
        self.create_with_access(dir_obj.fh, &name, mode, access)
            .await
    }

    pub(crate) async fn mkdir(
        &self,
        dir_fh: Bytes,
        dirname: &str,
        mode: u32,
    ) -> Result<mount::ObjRes> {
        // 不在 CREATE createattrs 传 mode，避免 NFS4ERR_ATTRNOTSUPP；创建后 SETATTR 设置。
        let bitmap = standard_getattr_bitmap();
        let resp = self
            .compound("mkdir", |b| {
                b.putfh(&dir_fh)
                    .create(2 /* NF4DIR */, dirname, &[], &[])
                    .getfh()
                    .getattr(&bitmap)
            })
            .await?;
        resp.op_ok(1)?; // PUTFH
        resp.op_ok(2)?; // CREATE
        // Log CREATE change_info
        if let Some(create_op) = resp.results.get(2) {
            let mut cdata = create_op.data.clone();
            if cdata.remaining() >= 20 {
                let atomic = cdata.get_u32() != 0;
                let before = cdata.get_u64();
                let after = cdata.get_u64();
                debug!(atomic, before, after, name = dirname, "CREATE change_info");
            }
        }
        let getfh = resp.op_ok(3)?;
        let mut fh_data = getfh.data.clone();
        let fh = decode_fh(&mut fh_data)?;
        let getattr = resp.op_ok(4)?;
        let mut attr_data = getattr.data.clone();
        let mut attr = decode_getattr_response(&mut attr_data)?;
        // 创建后通过 SETATTR 设置 mode，保持与 v3 行为一致。
        // 目录已创建成功，SETATTR 失败只记 warning，不影响整体结果。
        if let Err(e) = self
            .setattr(fh.clone(), None, Some(mode), None, None, None, None, None)
            .await
        {
            warn!(error = %e, mode, "mkdir: SETATTR mode failed after dir creation");
        } else {
            attr.file_mode = mode;
        }
        Ok(mount::ObjRes {
            fh,
            attr: Some(attr),
        })
    }

    pub(crate) async fn mkdir_path(&self, path: &str, mode: u32) -> Result<mount::ObjRes> {
        let (dir, name) = crate::split_path(path)?;
        let dir_obj = self.lookup_path(&dir).await?;
        self.mkdir(dir_obj.fh, &name, mode).await
    }
}