goosefs-sdk 0.2.0

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

//! Goosefs Master gRPC client for file system metadata operations.
//!
//! Wraps `FileSystemMasterClientService` (Master:9200) providing:
//! - `get_status` — stat / head
//! - `list_status` — list directory (client-side recursion when requested)
//! - `create_file` — create a new file
//! - `complete_file` — mark file write complete (with idempotency operation-ID)
//! - `remove_blocks` — clean up block metadata for in-flight or failed writes
//! - `delete` / `delete_with_options` — delete file or directory
//! - `rename` — rename / move
//! - `create_directory` — mkdir -p
//!
//! ## HA / Multi-Master Support
//!
//! When multiple Master addresses are configured, [`MasterClient::connect`]
//! uses [`MasterInquireClient`] to discover the Primary Master before
//! establishing the gRPC channel. If an RPC fails with a retriable error
//! (`Unavailable`, `DeadlineExceeded`), the client will re-discover the
//! Primary and rebuild the channel automatically.

use std::collections::{HashSet, VecDeque};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

use arc_swap::ArcSwap;
use tonic::service::interceptor::InterceptedService;
use tonic::transport::Channel;
use tracing::{debug, instrument, warn};

use crate::auth::{ChannelAuthenticator, ChannelIdInterceptor, SaslStreamGuard};
use crate::client::master_inquire::{create_master_inquire_client, MasterInquireClient};
use crate::config::GoosefsConfig;
use crate::error::{Error, Result};
use crate::fs::options::DeleteOptions;
use crate::metrics::registry::Counter;
use crate::proto::grpc::file::{
    file_system_master_client_service_client::FileSystemMasterClientServiceClient,
    CommitLocationPOptions, CommitLocationPRequest, CompleteFilePOptions, CompleteFilePRequest,
    CreateDirectoryPOptions, CreateDirectoryPRequest, CreateFilePOptions, CreateFilePRequest,
    DeletePOptions, DeletePRequest, FileInfo, FileSystemMasterCommonPOptions, FsOpPId,
    GetStatusPOptions, GetStatusPRequest, ListStatusPOptions, ListStatusPRequest,
    LoadMetadataPType, RemoveBlocksPRequest, RenamePOptions, RenamePRequest,
    ScheduleAsyncPersistencePOptions, ScheduleAsyncPersistencePRequest,
};
use crate::proto::grpc::{Bits, PMode};
use crate::proto::proto::shared::FileLocation;

/// Maximum number of RPC-level retries on retriable errors before giving up.
const MAX_RPC_RETRIES: u32 = 2;

/// Type alias for the authenticated gRPC client.
///
/// Both NOSASL and SIMPLE modes use `InterceptedService` wrapping;
/// the difference is that NOSASL skips the SASL handshake but still injects a channel-id.
type AuthenticatedFsClient =
    FileSystemMasterClientServiceClient<InterceptedService<Channel, ChannelIdInterceptor>>;

/// Immutable snapshot of the authenticated channel state.
///
/// `client` (which holds the tonic `Channel` + `channel-id` interceptor) and
/// `sasl_guard` (which keeps the SASL session alive on the Master side)
/// **must** travel together as a single unit: the Master uses the
/// `channel-id` injected by the interceptor to look up the SASL session,
/// so a stale `sasl_guard` paired with a fresh `client` (or vice versa) would
/// break authentication.
///
/// This struct enforces that pairing in the type system: it is never mutated
/// in place — instead, `MasterClient::reconnect` builds a brand-new
/// `AuthedState` and atomically swaps the `Arc` via `ArcSwap::store`. The old
/// `Arc<AuthedState>` is dropped only after the last in-flight reader
/// releases it, so the old SASL stream cannot be closed while anyone is still
/// using the old channel.
///
///
/// full consistency rationale.
struct AuthedState {
    client: AuthenticatedFsClient,
    /// Holds the SASL stream alive for the lifetime of `client`. `Option`
    /// because the test-only `from_channel` constructor and NOSASL mode do
    /// not need a SASL session.
    _sasl_guard: Option<SaslStreamGuard>,
}

/// Default mode for directories: 0755 (rwxr-xr-x)
pub fn default_dir_mode() -> PMode {
    PMode {
        owner_bits: Bits::All as i32,         // rwx
        group_bits: Bits::ReadExecute as i32, // r-x
        other_bits: Bits::ReadExecute as i32, // r-x
    }
}

/// Default mode for files: 0644 (rw-r--r--)
pub fn default_file_mode() -> PMode {
    PMode {
        owner_bits: Bits::ReadWrite as i32, // rw-
        group_bits: Bits::Read as i32,      // r--
        other_bits: Bits::Read as i32,      // r--
    }
}

/// Write-path-driven fields of `CompleteFilePOptions`.
///
/// # Java authority
///
/// Mirrors the builder calls in `GoosefsFileOutStream.close()`. Grouping them
/// keeps [`MasterClient::complete_file_with_options`] to two arguments as the
/// option set grows.
#[derive(Debug, Default, Clone)]
pub struct CompleteFileOptions {
    /// Final file length (`setUfsLength`). Java sends this for every write
    /// type, not just the persisting ones.
    pub ufs_length: Option<i64>,
    /// Idempotency token, carried in `commonOptions.operationId`.
    pub operation_id: Option<FsOpPId>,
    /// Last block's replica locations. ASYNC_THROUGH only — sending these for
    /// other write types makes the Master treat the file as persist-scheduled.
    pub locations: Vec<FileLocation>,
    /// Ask the Master to schedule an async-persist job.
    ///
    /// Mutually exclusive with [`force_persisted`](Self::force_persisted): a
    /// file that is already on the UFS must not also be queued for persisting.
    pub async_persist_options: Option<ScheduleAsyncPersistencePOptions>,
    /// Mark the file as already persisted.
    ///
    /// Set when the client degraded to a UFS-only write and that write
    /// completed, so the bytes are on the UFS before `CompleteFile` runs. The
    /// Master then fetches the UFS fingerprint and stamps the inode
    /// `PERSISTED` instead of queueing a persist job.
    pub force_persisted: Option<bool>,
}

/// Strip a trailing slash except for the filesystem root (`"/"`).
///
/// GooseFS listings generally omit trailing slashes, but treating `/foo` and
/// `/foo/` as distinct nodes would re-queue the same directory during BFS.
fn list_status_path_key(path: &str) -> &str {
    if path.len() > 1 {
        path.strip_suffix('/').unwrap_or(path)
    } else {
        path
    }
}

/// Child directory a recursive `list_status` BFS should visit next.
///
/// GooseFS `listStatus` normally returns children only. If a server echoes
/// the listed directory itself (`child == cur`, including `/foo` vs `/foo/`),
/// re-queueing it would loop forever. Empty paths are skipped.
fn list_status_bfs_child_dir<'a>(
    cur: &str,
    child_path: Option<&'a str>,
    is_folder: bool,
) -> Option<&'a str> {
    if !is_folder {
        return None;
    }
    let child = child_path.filter(|p| !p.is_empty())?;
    if list_status_path_key(child) == list_status_path_key(cur) {
        return None;
    }
    Some(child)
}

/// Split a UUID into the `FsOpPId` wire layout Java uses
/// (`UUID.getMostSignificantBits` / `getLeastSignificantBits`).
pub(crate) fn new_fs_op_id() -> FsOpPId {
    let (high, low) = uuid::Uuid::new_v4().as_u64_pair();
    FsOpPId {
        most_significant_bits: Some(high as i64),
        least_significant_bits: Some(low as i64),
    }
}

/// Java `FileSystemOptions.commonDefaults(conf, withOpId)`.
///
/// Master `Context.create` does not merge server defaults. Mutating RPCs
/// (create / delete / rename) pass a fresh `operation_id` generated **once
/// per call** and reused across `with_retry` attempts so Master exactly-once
/// semantics work. Read RPCs pass `None`.
pub(crate) fn common_p_options(
    sync_interval_ms: i64,
    operation_id: Option<FsOpPId>,
) -> FileSystemMasterCommonPOptions {
    FileSystemMasterCommonPOptions {
        sync_interval_ms: Some(sync_interval_ms),
        operation_id,
    }
}

/// `commonDefaults(conf, true)` — sync interval plus a fresh operation id.
fn write_common_p_options(sync_interval_ms: i64) -> FileSystemMasterCommonPOptions {
    common_p_options(sync_interval_ms, Some(new_fs_op_id()))
}

/// Java `FileSystemOptions.renameDefaults`: `persist` from
/// `goosefs.user.file.persist.on.rename` (default `false`) plus write-path
/// `commonOptions`.
pub(crate) fn rename_p_options(sync_interval_ms: i64, persist: bool) -> RenamePOptions {
    RenamePOptions {
        common_options: Some(write_common_p_options(sync_interval_ms)),
        persist: Some(persist),
    }
}

/// Fill missing `common_options` fields on a mutating RPC, matching Java
/// `defaults.mergeFrom(caller)`: caller-set values win, unset ones take the
/// config default / a fresh operation id.
fn fill_write_common_options(
    slot: &mut Option<FileSystemMasterCommonPOptions>,
    sync_interval_ms: i64,
) {
    match slot {
        None => *slot = Some(write_common_p_options(sync_interval_ms)),
        Some(existing) => {
            if existing.sync_interval_ms.is_none() {
                existing.sync_interval_ms = Some(sync_interval_ms);
            }
            if existing.operation_id.is_none() {
                existing.operation_id = Some(new_fs_op_id());
            }
        }
    }
}

/// Java `FileSystemOptions.getStatusDefaults` wire shape.
///
/// Master `GetStatusContext.create` does **not** merge server defaults. An
/// unset `load_metadata_type` is protobuf enum `0` (`NEVER`), which makes
/// `checkLoadMetadataOptions` reject UFS-only paths with
/// `Path "..." does not exist.` OpenDAL `stat` / `get_status` must therefore
/// always send `ONCE` (config default) unless the caller overrides.
pub(crate) fn get_status_p_options(
    load_metadata_type: Option<LoadMetadataPType>,
    sync_interval_ms: Option<i64>,
) -> GetStatusPOptions {
    GetStatusPOptions {
        load_metadata_type: load_metadata_type.map(|t| t as i32),
        common_options: sync_interval_ms.map(|ms| common_p_options(ms, None)),
        ..Default::default()
    }
}

/// Build `ListStatusPOptions` matching Java `FileSystemOptions.listStatusDefaults`.
///
/// Same Master-side hazard as [`get_status_p_options`]: `ListStatusContext.create`
/// does not merge server defaults, and `DefaultFileSystemMaster.listStatus` reads
/// `getLoadMetadataType()` without a `hasLoadMetadataType()` fallback. An unset
/// value is protobuf enum `0` (`NEVER`), which both forces `loadDescendantType`
/// to `NONE` and makes `checkLoadMetadataOptions` reject a UFS-only directory
/// with `Path "..." does not exist.` — so OpenDAL `list` would silently miss
/// COS objects that are not in the inode tree yet.
pub(crate) fn list_status_p_options(
    load_metadata_type: Option<LoadMetadataPType>,
    sync_interval_ms: Option<i64>,
) -> ListStatusPOptions {
    ListStatusPOptions {
        load_metadata_type: load_metadata_type.map(|t| t as i32),
        common_options: sync_interval_ms.map(|ms| common_p_options(ms, None)),
        ..Default::default()
    }
}

/// Client for Goosefs `FileSystemMasterClientService` (Master:9200).
///
/// In HA mode, the client holds a reference to the [`MasterInquireClient`]
/// and can automatically re-discover the Primary Master when RPCs fail.
///
/// ## Authentication
///
/// The client supports NOSASL and SIMPLE authentication modes.
/// When `config.auth_type` is `Simple`, the client performs a SASL PLAIN
/// handshake after establishing the gRPC channel, then injects a `channel-id`
/// metadata header into all subsequent RPCs.
///
/// ## Concurrency model
///
/// `state` is an [`ArcSwap`] holding the immutable
/// `(channel + sasl_guard)` pair. The RPC hot path uses
/// `state.load()` — a wait-free single atomic load — to obtain a snapshot,
/// then clones the lightweight `AuthenticatedFsClient` (which is itself an
/// `Arc`-shared tonic `Channel`).  Failover (`reconnect`) atomically
/// publishes a new snapshot via `state.store(...)`; readers either see the
/// old snapshot (still valid for in-flight requests) or the new one — never
/// a torn mix.
///
/// The hot-path counters (`counter_*`) are cached as `Arc<Counter>` here
/// **outside** of `AuthedState` on purpose: they are process-level metric
/// handles that must outlive any `reconnect` and must not be re-resolved
/// from the global `DashMap` on every RPC. See
///
/// rule.
#[derive(Clone)]
pub struct MasterClient {
    /// Atomically-swappable authenticated state (channel + SASL guard).
    state: Arc<ArcSwap<AuthedState>>,
    config: GoosefsConfig,
    inquire_client: Arc<dyn MasterInquireClient>,
    /// Per-client in-flight RPC counter, shared across all clones of this
    /// `MasterClient` (the `Arc` makes `#[derive(Clone)]` produce a shared
    /// counter rather than independent ones). Incremented in `with_retry`
    /// on entry and decremented on exit (success, error, or panic via the
    /// RAII guard). The `MasterClientPool::pick` P2C scheduler reads this
    /// to pick the least-loaded channel — crucially this count is accurate
    /// even for `MasterClient`s cloned out of the pool (e.g. by
    /// `GoosefsFileWriter`), which the previous pool-level counter could
    /// not track.
    inflight: Arc<AtomicUsize>,
    // ── Cached metric handles (lifetime-aligned with the MasterClient, not
    //    with any single channel/SASL session — see ). Caching avoids
    //    `crate::metrics::counter(name)` DashMap lookups on every RPC.
    counter_get_status_ops: Arc<Counter>,
    counter_get_status_latency_us: Arc<Counter>,
    counter_list_status_ops: Arc<Counter>,
    counter_list_status_latency_us: Arc<Counter>,
    counter_create_file_ops: Arc<Counter>,
    counter_create_dir_ops: Arc<Counter>,
    counter_delete_ops: Arc<Counter>,
    counter_rename_ops: Arc<Counter>,
    counter_rpc_errors_total: Arc<Counter>,
    counter_rpc_auth_errors: Arc<Counter>,
    counter_rpc_unavailable_errors: Arc<Counter>,
}

impl MasterClient {
    /// Connect to the Goosefs Master.
    ///
    /// In single-master mode, connects directly to `config.master_addr`.
    /// In HA mode (multiple addresses in `config.master_addrs`), uses
    /// [`PollingMasterInquireClient`](crate::client::master_inquire::PollingMasterInquireClient)
    /// to discover the Primary first.
    ///
    /// Authentication is performed according to `config.auth_type`.
    pub async fn connect(config: &GoosefsConfig) -> Result<Self> {
        let inquire_client = create_master_inquire_client(config);
        Self::connect_with_inquire(config, inquire_client).await
    }

    /// Connect using an externally-provided [`MasterInquireClient`].
    ///
    /// This is useful when sharing a single inquire client across multiple
    /// client types (e.g. `MasterClient` + `WorkerManagerClient`).
    pub async fn connect_with_inquire(
        config: &GoosefsConfig,
        inquire_client: Arc<dyn MasterInquireClient>,
    ) -> Result<Self> {
        let primary_addr = inquire_client.get_primary_rpc_address().await?;
        let (client, sasl_guard) = Self::build_authenticated_client(config, &primary_addr).await?;
        debug!(addr = %primary_addr, auth_type = %config.auth_type, "connected to Goosefs Master");

        Ok(Self::from_parts(
            AuthedState {
                client,
                _sasl_guard: sasl_guard,
            },
            config.clone(),
            inquire_client,
        ))
    }

    /// Internal constructor that wires up the `ArcSwap<AuthedState>` and
    /// caches the hot-path metric handles in one place. Both
    /// [`Self::connect_with_inquire`] and [`Self::from_channel`] go through
    /// this so the field-list stays single-sourced.
    fn from_parts(
        state: AuthedState,
        config: GoosefsConfig,
        inquire_client: Arc<dyn MasterInquireClient>,
    ) -> Self {
        Self {
            state: Arc::new(ArcSwap::from_pointee(state)),
            config,
            inquire_client,
            inflight: Arc::new(AtomicUsize::new(0)),
            counter_get_status_ops: crate::metrics::counter(
                crate::metrics::name::CLIENT_GET_STATUS_OPS,
            ),
            counter_get_status_latency_us: crate::metrics::counter(
                crate::metrics::name::CLIENT_GET_STATUS_LATENCY_US,
            ),
            counter_list_status_ops: crate::metrics::counter(
                crate::metrics::name::CLIENT_LIST_STATUS_OPS,
            ),
            counter_list_status_latency_us: crate::metrics::counter(
                crate::metrics::name::CLIENT_LIST_STATUS_LATENCY_US,
            ),
            counter_create_file_ops: crate::metrics::counter(
                crate::metrics::name::CLIENT_CREATE_FILE_OPS,
            ),
            counter_create_dir_ops: crate::metrics::counter(
                crate::metrics::name::CLIENT_CREATE_DIR_OPS,
            ),
            counter_delete_ops: crate::metrics::counter(crate::metrics::name::CLIENT_DELETE_OPS),
            counter_rename_ops: crate::metrics::counter(crate::metrics::name::CLIENT_RENAME_OPS),
            counter_rpc_errors_total: crate::metrics::counter(
                crate::metrics::name::CLIENT_RPC_ERRORS_TOTAL,
            ),
            counter_rpc_auth_errors: crate::metrics::counter(
                crate::metrics::name::CLIENT_RPC_AUTH_ERRORS,
            ),
            counter_rpc_unavailable_errors: crate::metrics::counter(
                crate::metrics::name::CLIENT_RPC_UNAVAILABLE_ERRORS,
            ),
        }
    }

    /// Create from an existing tonic channel (useful for testing / channel sharing).
    ///
    /// **Note**: This bypasses authentication. The channel is wrapped with a
    /// no-op channel-id interceptor for API compatibility.
    pub fn from_channel(channel: Channel, config: GoosefsConfig) -> Self {
        let inquire_client = create_master_inquire_client(&config);
        let interceptor = ChannelIdInterceptor::new("test-no-auth".to_string());
        let intercepted = InterceptedService::new(channel, interceptor);
        Self::from_parts(
            AuthedState {
                client: FileSystemMasterClientServiceClient::new(intercepted),
                _sasl_guard: None,
            },
            config,
            inquire_client,
        )
    }

    /// Build a gRPC channel and perform authentication, returning an authenticated client
    /// and the SASL stream guard that must be kept alive.
    async fn build_authenticated_client(
        config: &GoosefsConfig,
        addr: &str,
    ) -> Result<(AuthenticatedFsClient, Option<SaslStreamGuard>)> {
        let channel = Self::build_raw_channel(config, addr).await?;

        // Perform SASL authentication based on the configured auth type
        let authenticator = ChannelAuthenticator::new(
            config.auth_type,
            config.auth_username.clone(),
            None, // impersonation_user: not yet supported
        )
        .with_auth_timeout(config.auth_timeout);

        let mut auth_channel = authenticator.authenticate(channel).await?;
        let sasl_guard = auth_channel.take_sasl_guard();

        Ok((
            FileSystemMasterClientServiceClient::new(auth_channel.channel),
            sasl_guard,
        ))
    }

    /// Build a raw gRPC channel to a specific master address (without authentication).
    async fn build_raw_channel(config: &GoosefsConfig, addr: &str) -> Result<Channel> {
        let endpoint_uri = format!("http://{}", addr);
        let endpoint = Channel::from_shared(endpoint_uri)
            .map_err(|e| Error::ConfigError {
                message: format!("invalid master endpoint: {}", e),
            })?
            .connect_timeout(config.connect_timeout)
            .timeout(config.request_timeout);

        let channel = endpoint.connect().await?;
        Ok(channel)
    }

    /// Reconnect to the Primary Master after a failover.
    ///
    /// Resets the cached Primary in the inquire client, re-discovers the
    /// new Primary, rebuilds the gRPC channel, and re-authenticates.
    ///
    /// The new `(client, sasl_guard)` pair is published as a single
    /// [`AuthedState`] via [`ArcSwap::store`], so concurrent readers always
    /// observe a self-consistent snapshot. The old `Arc<AuthedState>` —
    /// containing the previous `sasl_guard` — is kept alive by any in-flight
    /// reader holding the old `Guard`, and is only dropped after the last
    /// such reader releases it. This guarantees that the old SASL stream is
    /// not closed while old-channel requests are still in flight.
    async fn reconnect(&self) -> Result<()> {
        // Reset cached primary so the inquire client re-polls all addresses.
        self.inquire_client.reset_cached_primary().await;

        let primary_addr = self.inquire_client.get_primary_rpc_address().await?;
        let (client, sasl_guard) =
            Self::build_authenticated_client(&self.config, &primary_addr).await?;
        // Single atomic publish: callers either see the previous AuthedState
        // in its entirety, or the new one — never a torn `(new client, old
        // guard)` mix.
        self.state.store(Arc::new(AuthedState {
            client,
            _sasl_guard: sasl_guard,
        }));
        debug!(addr = %primary_addr, "reconnected to Goosefs Master after failover");
        Ok(())
    }

    /// Execute an RPC with automatic retry on retriable errors.
    ///
    /// On retriable failure, the client reconnects to a (potentially new)
    /// Primary Master and retries up to [`MAX_RPC_RETRIES`] times.
    ///
    /// Each RPC attempt is wrapped with an in-flight counter guard so the
    /// P2C scheduler in [`MasterClientPool::pick`] sees an accurate load
    /// even for `MasterClient`s cloned out of the pool (e.g. by
    /// `GoosefsFileWriter`). The counter is shared across clones via the
    /// `Arc<AtomicUsize>` field, and the RAII guard guarantees decrement
    /// on success, error, or future cancellation (task drop).
    async fn with_retry<F, Fut, T>(&self, op_name: &str, mut f: F) -> Result<T>
    where
        // `FnMut` (rather than `Fn`) lets callers move owned state (e.g. the
        // request `path: String`) into the closure on the *first* attempt and
        // only `clone()` it inside the closure when a retry is actually
        // needed.
        F: FnMut(AuthenticatedFsClient) -> Fut,
        Fut: std::future::Future<Output = Result<T>>,
    {
        let mut last_err: Option<Error> = None;

        for attempt in 0..=MAX_RPC_RETRIES {
            // For retry attempts (attempt > 0) we know the previous call hit
            // a retriable error, which usually means the channel is dead.
            // Reconnect *before* re-sending — sending on a stale channel
            // just burns `request_timeout` for no gain. If the reconnect
            // itself fails, skip this attempt (the next iteration will try
            // reconnect again) so we don't consume retries on a known-bad
            // connection.
            if attempt > 0 {
                if let Err(reconnect_err) = self.reconnect().await {
                    warn!(
                        op = op_name,
                        attempt = attempt + 1,
                        error = %reconnect_err,
                        "reconnect failed; will retry reconnect on next attempt"
                    );
                    last_err = Some(Error::Internal {
                        message: format!("master reconnect failed: {}", reconnect_err),
                        source: None,
                    });
                    continue;
                }
            }

            // Wait-free atomic load: replaces the previous
            // `RwLock::read().await` round-trip with a single `Acquire` load.
            // The cloned client shares the underlying `tonic::Channel`
            // (which itself is `Arc`-internally cloneable and Send+Sync), so
            // this is cheap.
            let client: AuthenticatedFsClient = self.state.load().client.clone();

            // Mark this RPC as in-flight for P2C load balancing. The guard
            // decrements on drop — covering success, error, and future
            // cancellation (task drop) paths uniformly.
            self.inflight.fetch_add(1, Ordering::Relaxed);
            let _inflight_guard = InflightGuard(&self.inflight);

            match f(client).await {
                Ok(result) => return Ok(result),
                Err(err) => {
                    // Instrument: count RPC errors (use the cached Arc<Counter>
                    // to avoid a DashMap lookup on every error path).
                    self.counter_rpc_errors_total.inc(1);
                    // Classify the error
                    if err.is_authentication_error() {
                        self.counter_rpc_auth_errors.inc(1);
                    } else if err.is_unavailable() {
                        self.counter_rpc_unavailable_errors.inc(1);
                    }

                    if err.is_retriable() && attempt < MAX_RPC_RETRIES {
                        warn!(
                            op = op_name,
                            attempt = attempt + 1,
                            max = MAX_RPC_RETRIES,
                            error = %err,
                            "retriable error; will reconnect and retry"
                        );
                        last_err = Some(err);
                    } else {
                        return Err(err);
                    }
                }
            }
            // `_inflight_guard` drops here on the retry path → fetch_sub.
        }

        Err(last_err.unwrap_or_else(|| Error::Internal {
            message: format!("{}: exhausted all retries", op_name),
            source: None,
        }))
    }

    /// Get the file/directory status (equivalent to `stat` / `head`).
    ///
    /// Sends Java `FileSystemOptions.getStatusDefaults`:
    /// `loadMetadataType` from [`GoosefsConfig::file_metadata_load_type`]
    /// (default `ONCE`) and `syncIntervalMs` from
    /// [`GoosefsConfig::file_metadata_sync_interval`] (default `-1`).
    ///
    /// The Master handler uses client options as-is (`GetStatusContext.create`,
    /// no server-side merge). An unset `loadMetadataType` is proto enum `0`
    /// (`NEVER`), so COS/UFS files not yet in the namespace would return
    /// `NotFound`. Matching Java is required for OpenDAL `stat`.
    #[instrument(skip(self), fields(path = %path))]
    pub async fn get_status(&self, path: &str) -> Result<FileInfo> {
        self.get_status_with_load_type(
            path,
            Some(self.config.file_metadata_load_type),
            Some(self.config.file_metadata_sync_interval),
        )
        .await
    }

    /// `GetStatus` with an explicit `load_metadata_type` / `sync_interval_ms`.
    ///
    /// # Java authority
    ///
    /// The `close()` recovery path in `GoosefsFileOutStream` re-reads the file
    /// with `LoadMetadataPType.ALWAYS` and `syncIntervalMs = 0` after a
    /// `completeFile` failure, forcing the Master to re-import metadata from
    /// the UFS copy that was already written.
    ///
    /// Unlike [`Self::get_status`] this deliberately bypasses no client-side
    /// cache of its own — callers wanting cache semantics should go through
    /// `FileSystem::get_status_with_options`.
    #[instrument(skip(self), fields(path = %path, ?load_metadata_type, ?sync_interval_ms))]
    pub async fn get_status_with_load_type(
        &self,
        path: &str,
        load_metadata_type: Option<LoadMetadataPType>,
        sync_interval_ms: Option<i64>,
    ) -> Result<FileInfo> {
        let start = std::time::Instant::now();
        // Allocate the owned path exactly once.
        //
        // The closure captures `path_owned: Option<String>` by `&mut`. On
        // the first attempt we `take()` (move) into the request — zero
        // additional allocation. On a retry attempt (rare) the `Option` is
        // empty, so we re-allocate from `path` (`&str`) one more time. Since
        // `with_retry` accepts `FnMut`, this pattern is sound.
        let options = get_status_p_options(load_metadata_type, sync_interval_ms);
        let mut path_owned: Option<String> = Some(path.to_string());
        let result = self
            .with_retry("get_status", |mut client| {
                let req_path = path_owned.take().unwrap_or_else(|| path.to_string());
                let options = options.clone();
                async move {
                    let req = GetStatusPRequest {
                        path: Some(req_path),
                        options: Some(options),
                        request_id: None,
                    };
                    client
                        .get_status(req)
                        .await?
                        .into_inner()
                        .file_info
                        .ok_or_else(|| Error::missing_field("file_info"))
                }
            })
            .await;
        self.counter_get_status_ops.inc(1);
        self.counter_get_status_latency_us
            .inc(start.elapsed().as_micros() as i64);
        result
    }

    /// List the contents of a directory. Returns all FileInfo entries.
    ///
    /// GooseFS 2.0 dropped `ListStatusPOptions.recursive`; each Master RPC
    /// returns a single directory level. When `recursive` is `true`, this
    /// method walks descendants **client-side** (BFS), matching Java
    /// `BaseFileSystem.iterateStatusInternal` and the historical SDK contract
    /// that `list_status(path, true)` returns the full subtree.
    ///
    /// Every level — recursive or not — sends Java
    /// `FileSystemOptions.listStatusDefaults`: `loadMetadataType` from
    /// [`GoosefsConfig::file_metadata_load_type`] (default `ONCE`) and
    /// `syncIntervalMs` from [`GoosefsConfig::file_metadata_sync_interval`]
    /// (default `-1`). Callers that need `Never` / `Always` should use
    /// [`Self::list_status_with_load_type`]. The BFS skips a child whose path
    /// is the directory currently being listed (`/foo` and `/foo/` are the
    /// same node), so a server that echoes `self` cannot loop forever.
    ///
    /// Each level wraps a **server-side streaming** RPC — the server sends
    /// multiple `ListStatusPResponse` messages, each containing a batch
    /// of `FileInfo`.
    #[instrument(skip(self), fields(path = %path, recursive))]
    pub async fn list_status(&self, path: &str, recursive: bool) -> Result<Vec<FileInfo>> {
        self.list_status_with_load_type(path, recursive, None).await
    }

    /// List a directory, applying `load_metadata_type` on every Master RPC.
    ///
    /// `None` resolves to [`GoosefsConfig::file_metadata_load_type`] (default
    /// `ONCE`), and `syncIntervalMs` comes from the config — same as
    /// [`Self::list_status`].
    #[instrument(skip(self), fields(path = %path, recursive, ?load_metadata_type))]
    pub async fn list_status_with_load_type(
        &self,
        path: &str,
        recursive: bool,
        load_metadata_type: Option<LoadMetadataPType>,
    ) -> Result<Vec<FileInfo>> {
        self.list_status_with_options(
            path,
            recursive,
            load_metadata_type,
            Some(self.config.file_metadata_sync_interval),
        )
        .await
    }

    /// `ListStatus` with an explicit `load_metadata_type` / `sync_interval_ms`.
    ///
    /// Mirrors [`Self::get_status_with_load_type`] so that a `FileSystem`
    /// layer which resolved both values from `ListStatusOptions` can pass
    /// them straight through.
    #[instrument(
        skip(self),
        fields(path = %path, recursive, ?load_metadata_type, ?sync_interval_ms)
    )]
    pub async fn list_status_with_options(
        &self,
        path: &str,
        recursive: bool,
        load_metadata_type: Option<LoadMetadataPType>,
        sync_interval_ms: Option<i64>,
    ) -> Result<Vec<FileInfo>> {
        let load = load_metadata_type.unwrap_or(self.config.file_metadata_load_type);
        let options = list_status_p_options(Some(load), sync_interval_ms);
        if !recursive {
            return self.list_status_one_level(path, options).await;
        }

        // Client-side BFS: Master no longer accepts a recursive ListStatus option.
        let mut out: Vec<FileInfo> = Vec::new();
        let mut queue: VecDeque<String> = VecDeque::new();
        let mut visited: HashSet<String> = HashSet::new();
        let start = list_status_path_key(path).to_string();
        visited.insert(start.clone());
        queue.push_back(start);
        while let Some(cur) = queue.pop_front() {
            let items = self.list_status_one_level(&cur, options.clone()).await?;
            for fi in items {
                if let Some(child) =
                    list_status_bfs_child_dir(&cur, fi.path.as_deref(), fi.folder.unwrap_or(false))
                {
                    let key = list_status_path_key(child).to_string();
                    if visited.insert(key.clone()) {
                        queue.push_back(key);
                    }
                }
                out.push(fi);
            }
        }
        Ok(out)
    }

    /// One-level `ListStatus` RPC (no client-side descent).
    async fn list_status_one_level(
        &self,
        path: &str,
        options: ListStatusPOptions,
    ) -> Result<Vec<FileInfo>> {
        let start = std::time::Instant::now();
        let path = path.to_string();
        let result = self
            .with_retry("list_status", |mut client| {
                let path = path.clone();
                let options = options.clone();
                async move {
                    let req = ListStatusPRequest {
                        path: Some(path),
                        options: Some(options),
                        request_id: None,
                    };
                    let mut stream = client.list_status(req).await?.into_inner();
                    let mut result = Vec::new();
                    while let Some(resp) = stream.message().await? {
                        result.extend(resp.file_infos);
                    }
                    Ok(result)
                }
            })
            .await;
        self.counter_list_status_ops.inc(1);
        self.counter_list_status_latency_us
            .inc(start.elapsed().as_micros() as i64);
        result
    }

    /// Create a new file. Returns the `FileInfo` of the created file.
    #[instrument(skip(self, options), fields(path = %path))]
    pub async fn create_file(
        &self,
        path: &str,
        mut options: CreateFilePOptions,
    ) -> Result<FileInfo> {
        fill_write_common_options(
            &mut options.common_options,
            self.config.file_metadata_sync_interval,
        );
        let path = path.to_string();
        let result = self
            .with_retry("create_file", |mut client| {
                let path = path.clone();
                let options = options.clone();
                async move {
                    let req = CreateFilePRequest {
                        path: Some(path),
                        options: Some(options),
                    };
                    let resp = client.create_file(req).await?;
                    resp.into_inner()
                        .file_info
                        .ok_or_else(|| Error::missing_field("file_info"))
                }
            })
            .await;
        self.counter_create_file_ops.inc(1);
        result
    }

    /// Mark a file as completed (called after all blocks are written).
    ///
    /// # Idempotent operation ID
    ///
    /// `operation_id` is used by the Master for exactly-once semantics: if the
    /// RPC is retried after a network hiccup the Master detects the duplicate
    /// via `FsOpPId` and returns success without applying the operation twice.
    ///
    /// The caller (`GoosefsFileWriter`) generates a fresh `uuid::Uuid` at
    /// construction time and reuses it across all `complete_file` calls for the
    /// same write session.  The UUID is split into two `i64` halves via
    /// `Uuid::as_u64_pair()`:
    ///
    /// ```text
    /// (high, low) = uuid.as_u64_pair()
    /// FsOpPId { most_significant_bits: high as i64,
    ///           least_significant_bits: low  as i64 }
    /// ```
    ///
    /// This matches Java `UUID.getMostSignificantBits()` / `getLeastSignificantBits()`
    /// as verified in `DefaultFileSystemMaster.completeFile()`.
    ///
    /// # Note on Go SDK bug
    ///
    /// The Go SDK `base_filesystem.go:394-400` accepts an `operationID` parameter
    /// but **never writes it to the proto request**.  The Rust implementation
    /// fixes this: `operation_id` is always wired into `CompleteFilePOptions`.
    #[instrument(skip(self), fields(path = %path))]
    pub async fn complete_file(
        &self,
        path: &str,
        ufs_length: Option<i64>,
        operation_id: Option<FsOpPId>,
    ) -> Result<()> {
        self.complete_file_with_options(
            path,
            CompleteFileOptions {
                ufs_length,
                operation_id,
                ..Default::default()
            },
        )
        .await
    }

    /// `CompleteFile` with ASYNC_THROUGH location metadata and persist options.
    ///
    /// # Java authority
    ///
    /// Matches `GoosefsFileOutStream.close()` which puts the last block's
    /// `writeSucceedWorkers` into `CompleteFilePOptions.locations` and, for
    /// ASYNC_THROUGH, embeds `asyncPersistOptions` rather than issuing a
    /// separate `ScheduleAsyncPersistence` RPC.
    #[instrument(
        skip(self, opts),
        fields(path = %path, location_count = opts.locations.len())
    )]
    pub async fn complete_file_with_options(
        &self,
        path: &str,
        opts: CompleteFileOptions,
    ) -> Result<()> {
        let path = path.to_string();
        let sync_interval_ms = self.config.file_metadata_sync_interval;
        self.with_retry("complete_file", |mut client| {
            let path = path.clone();
            let opts = opts.clone();
            async move {
                let common_options = Some(common_p_options(sync_interval_ms, opts.operation_id));
                let req = CompleteFilePRequest {
                    path: Some(path),
                    options: Some(CompleteFilePOptions {
                        ufs_length: opts.ufs_length,
                        common_options,
                        locations: opts.locations,
                        async_persist_options: opts.async_persist_options,
                        force_persisted: opts.force_persisted,
                        ..Default::default()
                    }),
                    inode_id: None,
                };
                client.complete_file(req).await?;
                Ok(())
            }
        })
        .await
    }

    /// Commit a completed block's replica locations to Master.
    ///
    /// # Java authority
    ///
    /// Matches `GoosefsFileOutStream.commitCurrentBlock()` →
    /// `FileSystemMasterClient.commitLocation`. Called after every block
    /// except the last (the last block's locations travel with `completeFile`).
    #[instrument(skip(self, locations), fields(path = %path, block_id = block_id, location_count = locations.len()))]
    pub async fn commit_location(
        &self,
        path: &str,
        inode_id: Option<i64>,
        block_id: i64,
        locations: Vec<FileLocation>,
    ) -> Result<()> {
        if locations.is_empty() {
            return Ok(());
        }
        let path = path.to_string();
        self.with_retry("commit_location", |mut client| {
            let path = path.clone();
            let locations = locations.clone();
            async move {
                let req = CommitLocationPRequest {
                    path: Some(path),
                    inode_id,
                    block_id: Some(block_id),
                    options: Some(CommitLocationPOptions { locations }),
                };
                client.commit_location(req).await?;
                Ok(())
            }
        })
        .await
    }

    // -----------------------------------------------------------------------
    // RemoveBlocks RPC
    // -----------------------------------------------------------------------

    /// Request the Master to free block metadata for the given block IDs.
    ///
    /// This is the preferred cleanup path for `GoosefsFileWriter::cancel()`:
    /// it removes only the block metadata on the Master without touching the
    /// file-system namespace entry (the INCOMPLETE inode).
    ///
    /// Falls back to `delete_with_options(unchecked=true)` when this RPC fails.
    ///
    /// # Java authority
    ///
    /// Matches `FileSystemMasterClientServiceHandler.removeBlocks()` →
    /// `DefaultFileSystemMaster.removeBlocks(blockIds)`.
    #[instrument(skip(self, block_ids), fields(block_count = block_ids.len()))]
    pub async fn remove_blocks(&self, block_ids: Vec<i64>) -> Result<()> {
        if block_ids.is_empty() {
            return Ok(());
        }
        let block_ids_clone = block_ids.clone();
        self.with_retry("remove_blocks", |mut client| {
            let block_ids = block_ids_clone.clone();
            async move {
                let req = RemoveBlocksPRequest { block_ids };
                client.remove_blocks(req).await?;
                Ok(())
            }
        })
        .await
    }

    // -----------------------------------------------------------------------
    // Delete with full DeleteOptions
    // -----------------------------------------------------------------------

    /// Delete a file or directory with fine-grained options.
    ///
    /// Prefer this over the legacy [`delete`](Self::delete) wrapper when you need
    /// `unchecked` or `goosefs_only` semantics.
    ///
    /// See [`DeleteOptions`] for field semantics and Java authority notes.
    #[instrument(skip(self, opts), fields(path = %path))]
    pub async fn delete_with_options(&self, path: &str, opts: DeleteOptions) -> Result<()> {
        let path = path.to_string();
        let common_options = Some(write_common_p_options(
            self.config.file_metadata_sync_interval,
        ));
        self.with_retry("delete_with_options", |mut client| {
            let path = path.clone();
            let opts = opts.clone();
            async move {
                let req = DeletePRequest {
                    path: Some(path),
                    options: Some(DeletePOptions {
                        recursive: Some(opts.recursive),
                        unchecked: Some(opts.unchecked),
                        goosefs_only: Some(opts.goosefs_only),
                        common_options,
                        ..Default::default()
                    }),
                };
                client.remove(req).await?;
                Ok(())
            }
        })
        .await
    }

    /// Delete a file or directory (simple recursive wrapper).
    ///
    /// For `unchecked` or `goosefs_only` deletion use [`delete_with_options`](Self::delete_with_options)
    /// directly.
    #[instrument(skip(self), fields(path = %path, recursive = %recursive))]
    pub async fn delete(&self, path: &str, recursive: bool) -> Result<()> {
        let result = self
            .delete_with_options(
                path,
                DeleteOptions {
                    recursive,
                    ..Default::default()
                },
            )
            .await;
        self.counter_delete_ops.inc(1);
        result
    }

    /// Rename (move) a file or directory.
    #[instrument(skip(self), fields(src = %src, dst = %dst))]
    pub async fn rename(&self, src: &str, dst: &str) -> Result<()> {
        let src = src.to_string();
        let dst = dst.to_string();
        let options = rename_p_options(
            self.config.file_metadata_sync_interval,
            self.config.file_persist_on_rename,
        );
        let result = self
            .with_retry("rename", |mut client| {
                let src = src.clone();
                let dst = dst.clone();
                async move {
                    let req = RenamePRequest {
                        path: Some(src),
                        dst_path: Some(dst),
                        options: Some(options),
                    };
                    client.rename(req).await?;
                    Ok(())
                }
            })
            .await;
        self.counter_rename_ops.inc(1);
        result
    }

    /// Create a directory (recursive by default).
    ///
    /// Sets a default mode of `0755` (rwxr-xr-x) so that the corresponding
    /// UFS directory created by Goosefs has usable permissions.
    #[instrument(skip(self), fields(path = %path))]
    pub async fn create_directory(&self, path: &str, recursive: bool) -> Result<()> {
        let path = path.to_string();
        // `allow_exists=true` is an intentional OpenDAL mkdir -p divergence
        // from Java `createDirectoryDefaults` (`allowExists=false`). commonOptions
        // still match Java: syncIntervalMs from config + a fresh operation id.
        let common_options = Some(write_common_p_options(
            self.config.file_metadata_sync_interval,
        ));
        let result = self
            .with_retry("create_directory", |mut client| {
                let path = path.clone();
                async move {
                    let req = CreateDirectoryPRequest {
                        path: Some(path),
                        options: Some(CreateDirectoryPOptions {
                            recursive: Some(recursive),
                            allow_exists: Some(true),
                            mode: Some(default_dir_mode()),
                            common_options,
                            ..Default::default()
                        }),
                    };
                    client.create_directory(req).await?;
                    Ok(())
                }
            })
            .await;
        self.counter_create_dir_ops.inc(1);
        result
    }

    /// Schedule asynchronous persistence for a file.
    /// This will persist the file to the underlying storage system.
    #[instrument(skip(self), fields(path = %path))]
    pub async fn schedule_async_persistence(
        &self,
        path: &str,
        persistence_wait_time: Option<i64>,
    ) -> Result<()> {
        let path = path.to_string();
        // Java `scheduleAsyncPersistDefaults` uses `commonDefaults(conf)` —
        // sync interval, no operation id.
        let common_options = Some(common_p_options(
            self.config.file_metadata_sync_interval,
            None,
        ));
        self.with_retry("schedule_async_persistence", |mut client| {
            let path = path.clone();
            async move {
                let req = ScheduleAsyncPersistencePRequest {
                    path: Some(path),
                    options: Some(ScheduleAsyncPersistencePOptions {
                        common_options,
                        persistence_wait_time,
                    }),
                };
                client.schedule_async_persistence(req).await?;
                Ok(())
            }
        })
        .await
    }

    /// Get a reference to the underlying config.
    pub fn config(&self) -> &GoosefsConfig {
        &self.config
    }

    /// Get a reference to the underlying inquire client.
    ///
    /// Useful for sharing the same inquire client with `WorkerManagerClient`.
    pub fn inquire_client(&self) -> &Arc<dyn MasterInquireClient> {
        &self.inquire_client
    }
}

// ── In-flight RPC counting (P2C load balancing) ─────────────────────────────

/// RAII guard that decrements a `MasterClient`'s in-flight RPC counter on drop.
///
/// Created in [`MasterClient::with_retry`] around each RPC attempt. The guard
/// is panic-safe and cancellation-safe: if the future is dropped before
/// completion (e.g. `tokio::task` cancellation), the guard's `Drop` still
/// runs, so the counter never leaks.
struct InflightGuard<'a>(&'a AtomicUsize);

impl Drop for InflightGuard<'_> {
    fn drop(&mut self) {
        self.0.fetch_sub(1, Ordering::Relaxed);
    }
}

// ── Master connection pool ────────────────────────────────────────────────

/// A pool of [`MasterClient`]s over independent HTTP/2 channels.
///
/// # Why
///
/// A single tonic [`Channel`] multiplexes all RPCs over one HTTP/2 connection,
/// which caps concurrency at `SETTINGS_MAX_CONCURRENT_STREAMS` (default 100).
/// Under 256-way concurrency over remote RTT the surplus requests queue in
/// `tower::Buffer`, which is the measured root cause of the remote GetFileStatus
/// / OpenFile regression vs Java (Java defaults to a channel pool). Spreading
/// requests across `master_connection_pool_size` channels removes the queue.
///
/// # Scheduling
///
/// Two strategies are available via `GoosefsConfig::master_connection_pool_schedule`:
///
/// - **RoundRobin** (default): cycles through channels in order. Wait-free,
///   zero overhead, no in-flight tracking required.
/// - **P2C**: Power of Two Choices — uniformly samples two distinct channels
///   with a fast PRNG (`fastrand`) and picks the one with fewer in-flight
///   RPCs. Per-channel in-flight counts are tracked inside each
///   [`MasterClient`] (incremented in `with_retry`, decremented on RPC
///   completion), so the count is accurate even for `MasterClient`s cloned
///   out of the pool (e.g. by `GoosefsFileWriter`).
///
/// # HA consistency
///
/// Every pooled client is constructed with the **same** `inquire_client`, so a
/// failover decision is shared: all channels re-discover and switch to the same
/// new Primary, eliminating split-brain. Each channel performs its own SASL
/// handshake and carries a unique `channel-id`, fully compatible with the
/// `ArcSwap<AuthedState>` model.
pub struct MasterClientPool {
    clients: Vec<Arc<MasterClient>>,
    schedule: crate::config::MasterPoolSchedule,
    /// Round-robin cursor for `RoundRobin` schedule (Relaxed, wait-free).
    rr: AtomicUsize,
}

impl MasterClientPool {
    /// Connect a pool of `config.master_connection_pool_size` master clients,
    /// all sharing the supplied `inquire_client`.
    ///
    /// The size is clamped to at least 1, so this is a strict superset of the
    /// previous single-channel behaviour (`size = 1`).
    pub async fn connect_with_inquire(
        config: &GoosefsConfig,
        inquire_client: Arc<dyn MasterInquireClient>,
    ) -> Result<Self> {
        let size = config.master_connection_pool_size.max(1);
        // Connect all channels concurrently to avoid multiplying cold-start
        // latency by `size` (each connect = TCP + SASL handshake). All
        // connections share the same inquire_client, which is safe under
        // concurrent access (it deduplicates primary discovery internally).
        let clients: Vec<Arc<MasterClient>> = futures::future::try_join_all((0..size).map(|_| {
            let config = config.clone();
            let inquire = inquire_client.clone();
            async move {
                MasterClient::connect_with_inquire(&config, inquire)
                    .await
                    .map(Arc::new)
            }
        }))
        .await?;
        debug!(pool_size = size, "MasterClientPool connected");
        Ok(Self {
            clients,
            schedule: config.master_connection_pool_schedule,
            rr: AtomicUsize::new(0),
        })
    }

    /// Pick the next client according to the configured scheduling strategy.
    ///
    /// - `RoundRobin` (default): cycle through channels in order. Wait-free,
    ///   zero overhead, no in-flight tracking required.
    /// - `P2C`: Power of Two Choices — uniformly sample two distinct channels
    ///   at random and select the one with fewer in-flight RPCs. The
    ///   per-channel in-flight count is maintained inside
    ///   [`MasterClient::with_retry`] (not by this method), so it stays
    ///   accurate even for clients cloned out of the pool.
    ///
    /// Returns `Arc<MasterClient>` — callers interact with it exactly as
    /// before. The in-flight counter lives inside `MasterClient` itself
    /// (shared via `Arc<AtomicUsize>` across clones), so P2C load awareness
    /// works regardless of whether the caller holds the `Arc` directly or
    /// clones the inner `MasterClient`.
    pub fn pick(&self) -> Arc<MasterClient> {
        let n = self.clients.len();
        if n == 1 {
            return self.clients[0].clone();
        }

        match self.schedule {
            crate::config::MasterPoolSchedule::RoundRobin => {
                let idx = self.rr.fetch_add(1, Ordering::Relaxed) % n;
                self.clients[idx].clone()
            }
            crate::config::MasterPoolSchedule::P2C => {
                // Uniformly sample two distinct indices with a fast PRNG.
                let a = fastrand::usize(0..n);
                let b = loop {
                    let b = fastrand::usize(0..n);
                    if b != a {
                        break b;
                    }
                };

                // Read per-client in-flight counts (tracked inside each
                // MasterClient via with_retry, so clone-out-of-pool RPCs
                // are counted too).
                let la = self.clients[a].inflight.load(Ordering::Relaxed);
                let lb = self.clients[b].inflight.load(Ordering::Relaxed);
                let idx = if la <= lb { a } else { b };

                // TOCTOU note: the counter is only incremented inside
                // with_retry (not at pick time). This leaves a sub-
                // microsecond window between pick() and the first RPC
                // increment, during which another pick() could observe
                // the same stale value. In practice this is negligible —
                // callers immediately .await on the returned client —
                // and a fetch_add here would leak one increment per
                // pick() (unbalanced against with_retry's +1/-1 per RPC).

                debug!(
                    a_idx = a,
                    b_idx = b,
                    a_inflight = la,
                    b_inflight = lb,
                    picked = idx,
                    "P2C pick"
                );

                self.clients[idx].clone()
            }
        }
    }

    /// Number of pooled channels.
    pub fn size(&self) -> usize {
        self.clients.len()
    }
}

#[cfg(test)]
mod tests {
    //! Concurrency-correctness tests for the `ArcSwap<AuthedState>`-based
    //! state model introduced as part of the GetFileStatus performance
    //! optimisation work.  See
    //!
    //! the rationale and the gating-test requirement.
    //!
    //! These tests intentionally do **not** spin up a real Master server.
    //! They exercise the *type-level* invariant that motivates the change:
    //!
    //! 1. The `(client, sasl_guard)` pair is published as a single
    //!    immutable `Arc<AuthedState>`.  A concurrent reader either sees
    //!    the previous publication in its entirety, or the new one — never
    //!    a torn `(new client, old guard)` mix.
    //!
    //! 2. The previous publication's resources (in particular the SASL
    //!    guard standing in for `SaslStreamGuard` here) are *not* dropped
    //!    until the last reader releases its `Arc`.  This is the
    //!    "old SASL stream cannot be closed while in-flight requests still
    //!    use the old channel" property.
    //!
    //! Both properties are checked against a stand-in payload struct that
    //! mirrors the shape of `AuthedState` (channel + guard).  The real
    //! `AuthedState` is private to the module and uses tonic's generated
    //! client stub which is hard to instantiate in a unit test, so the
    //! stand-in keeps the test focused on the `ArcSwap` semantics that
    //! `MasterClient` relies on.

    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
    use std::sync::{Arc, Barrier};
    use std::thread;
    use std::time::{Duration, Instant};

    use arc_swap::ArcSwap;

    #[test]
    fn list_status_bfs_skips_non_folders_and_empty_paths() {
        assert_eq!(
            super::list_status_bfs_child_dir("/a", Some("/a/b.bin"), false),
            None
        );
        assert_eq!(super::list_status_bfs_child_dir("/a", Some(""), true), None);
        assert_eq!(super::list_status_bfs_child_dir("/a", None, true), None);
    }

    #[test]
    fn list_status_bfs_skips_self_entry_to_avoid_infinite_loop() {
        assert_eq!(
            super::list_status_bfs_child_dir("/data", Some("/data"), true),
            None,
            "re-queueing the listed directory would loop forever"
        );
        assert_eq!(
            super::list_status_bfs_child_dir("/data", Some("/data/nested"), true),
            Some("/data/nested")
        );
    }

    #[test]
    fn list_status_path_key_strips_trailing_slash_except_root() {
        assert_eq!(super::list_status_path_key("/"), "/");
        assert_eq!(super::list_status_path_key(""), "");
        assert_eq!(super::list_status_path_key("/data"), "/data");
        assert_eq!(super::list_status_path_key("/data/"), "/data");
        assert_eq!(super::list_status_path_key("/data/nested/"), "/data/nested");
    }

    #[test]
    fn list_status_bfs_treats_trailing_slash_as_same_node() {
        assert_eq!(
            super::list_status_bfs_child_dir("/data", Some("/data/"), true),
            None
        );
        assert_eq!(
            super::list_status_bfs_child_dir("/data/", Some("/data"), true),
            None
        );
        assert_eq!(
            super::list_status_bfs_child_dir("/data/", Some("/data/nested/"), true),
            Some("/data/nested/")
        );
    }

    /// Stand-in for `AuthedState`.  The exact field types do not matter for
    /// the property we are testing — what matters is that:
    /// - `epoch` is an immutable per-publication tag (analogue of
    ///   "this channel's `channel-id`")
    /// - `guard` is a resource whose drop must be deferred until no reader
    ///   is using this snapshot any more (analogue of `SaslStreamGuard`).
    struct AuthedStateLike {
        /// Identifies which `store(...)` produced this snapshot.  In the
        /// real code, the tonic `Channel`'s `channel-id` plays the same
        /// role.
        epoch: u64,
        /// Same epoch as above, but inside a separately-allocated leaf —
        /// in the real code the SASL session id would live separately
        /// from the channel-id metadata.  Reading both and asserting they
        /// match is what proves there is no torn read.
        guard_epoch: Arc<u64>,
        /// When the snapshot is dropped, increments the shared counter.
        /// Mirrors `SaslStreamGuard`'s `Drop` impl (which would close the
        /// SASL stream on the Master side).
        drop_counter: Arc<AtomicUsize>,
    }

    impl Drop for AuthedStateLike {
        fn drop(&mut self) {
            self.drop_counter.fetch_add(1, Ordering::SeqCst);
        }
    }

    fn new_state(epoch: u64, drop_counter: Arc<AtomicUsize>) -> Arc<AuthedStateLike> {
        Arc::new(AuthedStateLike {
            epoch,
            guard_epoch: Arc::new(epoch),
            drop_counter,
        })
    }

    /// Property 1 — atomic publication.
    ///
    /// N reader threads continuously load the current `AuthedState` and
    /// assert that the `(epoch, guard_epoch)` pair is internally
    /// consistent.  Meanwhile a writer thread atomically swaps in fresh
    /// states.  A torn read would produce a snapshot whose `guard_epoch`
    /// disagrees with `epoch` — which can never happen with `ArcSwap`
    /// because the *whole* `Arc<AuthedState>` is replaced as one pointer.
    #[test]
    fn arcswap_publication_is_atomic_under_concurrent_readers() {
        const READERS: usize = 32;
        const RECONNECT_ROUNDS: usize = 200;

        let drop_counter = Arc::new(AtomicUsize::new(0));
        let state = Arc::new(ArcSwap::from(new_state(0, drop_counter.clone())));
        let stop = Arc::new(AtomicBool::new(false));
        // Wait for every reader (and the writer) to be scheduled before the
        // reconnect loop starts — otherwise a slow CI runner can finish all
        // rounds before some reader threads ever enter their loop, which
        // falsely fails with "reader observed nothing".
        let ready = Arc::new(Barrier::new(READERS + 1));

        let mut readers = Vec::with_capacity(READERS);
        for _ in 0..READERS {
            let state = state.clone();
            let stop = stop.clone();
            let ready = ready.clone();
            readers.push(thread::spawn(move || {
                ready.wait();
                let mut observed_epochs: Vec<u64> = Vec::new();
                while !stop.load(Ordering::Relaxed) {
                    let snap = state.load();
                    // The two fields are written by *different* allocations
                    // in different orders; only the `Arc<AuthedStateLike>`
                    // pointer publication is atomic.  This pair must
                    // always agree.
                    assert_eq!(
                        snap.epoch, *snap.guard_epoch,
                        "torn read: ArcSwap published a half-swapped snapshot",
                    );
                    observed_epochs.push(snap.epoch);
                }
                // One final load after stop so every reader records at least
                // the terminal published epoch even if scheduling was tight.
                let snap = state.load();
                assert_eq!(snap.epoch, *snap.guard_epoch);
                observed_epochs.push(snap.epoch);
                observed_epochs
            }));
        }

        ready.wait();

        // Writer: simulate `reconnect` events by store()'ing fresh states.
        for round in 1..=RECONNECT_ROUNDS {
            state.store(new_state(round as u64, drop_counter.clone()));
            // Yield so readers have a chance to observe each epoch.
            thread::sleep(Duration::from_micros(50));
        }

        stop.store(true, Ordering::Relaxed);
        // Make sure each reader saw at least one swap take effect.
        for r in readers {
            let observed = r.join().expect("reader thread panicked");
            assert!(!observed.is_empty(), "reader observed nothing");
            let max = observed.iter().copied().max().unwrap();
            assert!(
                max >= 1,
                "reader never saw a reconnect-published epoch (max={})",
                max,
            );
        }
    }

    /// Property 2 — no premature drop.
    ///
    /// A reader that has already `load()`ed a snapshot is then *paused*
    /// (e.g. parked between obtaining the channel and finishing the gRPC
    /// round-trip).  During the pause we run many more `store(...)`
    /// rounds.  The reader's `Guard`/`Arc<AuthedState>` keeps the old
    /// snapshot alive, so its `drop_counter` must NOT have ticked for
    /// that particular epoch yet.  Once the reader drops its handle, the
    /// old snapshot finally gets reclaimed.
    ///
    /// This is what guarantees, in the real code, that
    /// `SaslStreamGuard::drop` (which would close the SASL stream on the
    /// Master and unregister the `channel-id`) never fires while there
    /// are still in-flight RPCs holding a clone of the old client.
    #[test]
    fn old_snapshot_outlives_concurrent_swap_until_reader_releases() {
        let drop_counter = Arc::new(AtomicUsize::new(0));
        let state = Arc::new(ArcSwap::from(new_state(1, drop_counter.clone())));

        // Reader grabs a snapshot and *holds it*.
        let held = state.load_full();
        assert_eq!(held.epoch, 1);

        // While the reader is still holding `held`, simulate a flurry of
        // reconnect events.
        for round in 2..=50 {
            state.store(new_state(round, drop_counter.clone()));
        }

        // The held snapshot must still be alive, hence its drop counter
        // contribution has not fired.
        // Other (orphaned-on-store) snapshots may have been dropped — but
        // *not the one we hold*.  Verify by inspecting the held snapshot
        // directly: if it had been dropped we would not be able to read
        // its fields without UB; we additionally assert that the absolute
        // drop count is < total publications, i.e. at least one snapshot
        // (the one we hold) is still alive.
        let observed_drops = drop_counter.load(Ordering::SeqCst);
        // 50 publications happened (epochs 1..=50). The current one in
        // `state` and the one held by `held` must both still be alive →
        // at most 48 drops so far.
        assert!(
            observed_drops <= 48,
            "old snapshot was dropped while a reader still held it: \
             drops = {} (expected <= 48)",
            observed_drops,
        );
        assert_eq!(held.epoch, 1, "held snapshot was mutated in place");

        // Release the reader's hold.
        drop(held);

        // Replace the still-current snapshot too so that *no* live Arc
        // remains, then wait for ArcSwap's lazy reclamation to settle.
        state.store(new_state(999, drop_counter.clone()));
        let deadline = Instant::now() + Duration::from_secs(2);
        loop {
            // 50 original epochs + 1 final = 51 publications, but the
            // final-stored one is still in `state`, so we expect 50 drops.
            if drop_counter.load(Ordering::SeqCst) >= 50 {
                break;
            }
            if Instant::now() > deadline {
                panic!(
                    "expected >= 50 drops after releasing the held snapshot, \
                     observed {}",
                    drop_counter.load(Ordering::SeqCst),
                );
            }
            thread::sleep(Duration::from_millis(5));
        }
    }

    // ── P2C scheduler + in-flight counter tests ───────────────────────────
    //
    // These tests cover the P2C `pick()` selection logic and the
    // `InflightGuard` RAII semantics introduced to fix the "counter does
    // not track cloned MasterClient RPCs" issue. They do **not** perform any
    // network I/O — a lazy tonic channel is used so no connection is
    // established.

    use super::{InflightGuard, MasterClient, MasterClientPool};
    use crate::config::GoosefsConfig;
    use std::panic::AssertUnwindSafe;

    /// Build a `MasterClient` without any network I/O.
    ///
    /// `connect_lazy()` creates a `Channel` that only connects on the first
    /// RPC — which these tests never send. The resulting client has a valid
    /// `inflight: Arc<AtomicUsize>` field (initially 0) that the P2C
    /// scheduler reads.
    fn make_test_master_client() -> MasterClient {
        let endpoint = tonic::transport::Endpoint::from_static("http://localhost:0").connect_lazy();
        MasterClient::from_channel(endpoint, GoosefsConfig::new("localhost:0"))
    }

    /// Build a pool of `n` test clients with P2C scheduling (no network I/O).
    fn make_test_pool(n: usize) -> MasterClientPool {
        let clients: Vec<Arc<MasterClient>> = (0..n)
            .map(|_| Arc::new(make_test_master_client()))
            .collect();
        MasterClientPool {
            clients,
            schedule: crate::config::MasterPoolSchedule::P2C,
            rr: AtomicUsize::new(0),
        }
    }

    /// Under unequal load, `pick()` must always select the lighter channel.
    #[tokio::test]
    async fn pick_chooses_lighter_candidate() {
        let pool = make_test_pool(2);
        // Artificially load client 0 with 10 in-flight RPCs; client 1 stays idle.
        pool.clients[0].inflight.store(10, Ordering::Relaxed);
        // Regardless of which two candidates are sampled, the lighter one
        // (client 1) must win every time.
        for _ in 0..200 {
            let picked = pool.pick();
            assert!(
                Arc::ptr_eq(&picked, &pool.clients[1]),
                "pick() selected the heavier channel"
            );
        }
    }

    /// With `n >= 2`, `pick()` never compares a channel against itself —
    /// the two sampled indices are always distinct.
    #[tokio::test]
    async fn pick_samples_two_distinct_candidates() {
        // With n=1 there is no choice; with n=2 the only possible pair is
        // (0,1), so every pick observes both. Verify that over many picks
        // both channels are returned (proving the loop produces a != b).
        let pool = make_test_pool(2);
        let mut saw_0 = false;
        let mut saw_1 = false;
        // Set equal load so selection is driven purely by the PRNG +
        // tie-break (la <= lb → a), not by load asymmetry.
        for _ in 0..1000 {
            let picked = pool.pick();
            if Arc::ptr_eq(&picked, &pool.clients[0]) {
                saw_0 = true;
            } else if Arc::ptr_eq(&picked, &pool.clients[1]) {
                saw_1 = true;
            } else {
                panic!("pick() returned a client not in the pool");
            }
        }
        assert!(
            saw_0 && saw_1,
            "pick() never sampled one of the two channels"
        );
    }

    /// Under equal load, `pick()` distributes selections across all channels
    /// (no channel is starved over a reasonable sample size).
    #[tokio::test]
    async fn pick_balances_equal_loads() {
        let n = 4;
        let pool = make_test_pool(n);
        let mut hits = vec![0usize; n];
        for _ in 0..4000 {
            let picked = pool.pick();
            for (i, c) in pool.clients.iter().enumerate() {
                if Arc::ptr_eq(&picked, c) {
                    hits[i] += 1;
                    break;
                }
            }
        }
        // Every channel must receive at least one pick (P2C with uniform
        // sampling visits all channels given enough trials). We don't assert
        // a tight distribution — just non-starvation.
        for (i, &h) in hits.iter().enumerate() {
            assert!(h > 0, "channel {} was starved by pick()", i);
        }
    }

    /// `InflightGuard` decrements the counter on normal scope exit.
    #[test]
    fn inflight_counter_decrements_on_normal_exit() {
        let counter = AtomicUsize::new(0);
        {
            counter.fetch_add(1, Ordering::Relaxed);
            let _guard = InflightGuard(&counter);
            assert_eq!(
                counter.load(Ordering::Relaxed),
                1,
                "counter must be 1 while guard alive"
            );
        }
        assert_eq!(
            counter.load(Ordering::Relaxed),
            0,
            "counter must return to 0 after guard drops"
        );
    }

    /// `InflightGuard` decrements the counter on early (explicit) drop.
    #[test]
    fn inflight_counter_decrements_on_early_drop() {
        let counter = AtomicUsize::new(0);
        counter.fetch_add(1, Ordering::Relaxed);
        let guard = InflightGuard(&counter);
        assert_eq!(counter.load(Ordering::Relaxed), 1);
        drop(guard); // explicit early drop
        assert_eq!(
            counter.load(Ordering::Relaxed),
            0,
            "counter must be 0 after early drop"
        );
    }

    /// `InflightGuard` decrements the counter even when the containing
    /// stack unwinds due to a panic (cancellation safety).
    #[test]
    fn inflight_guard_decrements_on_panic() {
        let counter = Arc::new(AtomicUsize::new(0));
        let counter_for_unwind = counter.clone();
        let result = std::panic::catch_unwind(AssertUnwindSafe(move || {
            counter_for_unwind.fetch_add(1, Ordering::Relaxed);
            let _guard = InflightGuard(&counter_for_unwind);
            assert_eq!(
                counter_for_unwind.load(Ordering::Relaxed),
                1,
                "counter must be 1 while guard alive"
            );
            panic!("simulated panic mid-RPC");
        }));
        assert!(result.is_err(), "test should have panicked");
        assert_eq!(
            counter.load(Ordering::Relaxed),
            0,
            "guard must decrement on panic unwind (cancellation safety)"
        );
    }

    /// `MasterClient::clone()` shares the in-flight counter via `Arc`, so
    /// RPCs issued through a cloned client (e.g. by `GoosefsFileWriter`)
    /// are visible to the P2C scheduler. This is the core correctness
    /// property that fixes the "counter does not track cloned MasterClient"
    /// issue.
    #[tokio::test]
    async fn master_client_clone_shares_inflight_counter() {
        let original = make_test_master_client();
        let cloned = original.clone();
        // Mutate via the clone; the original must observe the same value.
        cloned.inflight.store(7, Ordering::Relaxed);
        assert_eq!(
            original.inflight.load(Ordering::Relaxed),
            7,
            "clone must share the in-flight counter (Arc<AtomicUsize>)"
        );
        // And vice-versa.
        original.inflight.store(3, Ordering::Relaxed);
        assert_eq!(
            cloned.inflight.load(Ordering::Relaxed),
            3,
            "mutations via original must be visible to clone"
        );
    }

    /// `pick()` returns `Arc<MasterClient>` — cloning the Arc shares the
    /// same underlying `MasterClient` (and thus the same in-flight counter).
    #[tokio::test]
    async fn pick_returns_arc_sharing_inflight() {
        let pool = make_test_pool(2);
        let picked = pool.pick();
        let cloned = Arc::clone(&picked);
        assert!(
            Arc::ptr_eq(&picked, &cloned),
            "Arc::clone must share the same MasterClient"
        );
    }

    /// Java `getStatusDefaults` always sets `loadMetadataType=ONCE` and
    /// `syncIntervalMs=-1`. An all-None options struct is proto `NEVER`.
    #[test]
    fn get_status_p_options_matches_java_defaults() {
        use crate::proto::grpc::file::LoadMetadataPType;

        let opts = super::get_status_p_options(Some(LoadMetadataPType::Once), Some(-1));
        assert_eq!(
            opts.load_metadata_type,
            Some(LoadMetadataPType::Once as i32)
        );
        assert_eq!(
            opts.common_options
                .as_ref()
                .and_then(|c| c.sync_interval_ms),
            Some(-1)
        );
    }

    #[test]
    fn get_status_p_options_unset_is_never_on_the_wire() {
        let opts = super::get_status_p_options(None, None);
        assert_eq!(
            opts.load_metadata_type, None,
            "unset load_metadata_type is proto NEVER (0) on the Master"
        );
        assert!(opts.common_options.is_none());
    }

    #[test]
    fn get_status_uses_config_load_type_and_sync_interval() {
        use crate::proto::grpc::file::LoadMetadataPType;

        let cfg = GoosefsConfig::new("localhost:0");
        assert_eq!(cfg.file_metadata_load_type, LoadMetadataPType::Once);
        assert_eq!(cfg.file_metadata_sync_interval, -1);
        let opts = super::get_status_p_options(
            Some(cfg.file_metadata_load_type),
            Some(cfg.file_metadata_sync_interval),
        );
        assert_eq!(opts.load_metadata_type, Some(1)); // ONCE
        assert_eq!(
            opts.common_options
                .as_ref()
                .and_then(|c| c.sync_interval_ms),
            Some(-1)
        );
    }

    /// Java `listStatusDefaults` sets the same two fields as
    /// `getStatusDefaults`, for recursive and non-recursive listings alike.
    #[test]
    fn list_status_p_options_matches_java_defaults() {
        use crate::proto::grpc::file::LoadMetadataPType;

        let opts = super::list_status_p_options(Some(LoadMetadataPType::Once), Some(-1));
        assert_eq!(
            opts.load_metadata_type,
            Some(LoadMetadataPType::Once as i32)
        );
        assert_eq!(
            opts.common_options
                .as_ref()
                .and_then(|c| c.sync_interval_ms),
            Some(-1)
        );
    }

    /// A non-recursive `list_status` used to send no `loadMetadataType` at
    /// all, which the Master reads as `NEVER` — OpenDAL `list` then missed
    /// UFS-only entries. Pin the resolved default instead.
    #[test]
    fn list_status_p_options_unset_is_never_on_the_wire() {
        let opts = super::list_status_p_options(None, None);
        assert_eq!(
            opts.load_metadata_type, None,
            "unset load_metadata_type is proto NEVER (0) on the Master"
        );
        assert!(opts.common_options.is_none());
    }

    #[test]
    fn list_status_uses_config_load_type_and_sync_interval() {
        use crate::proto::grpc::file::LoadMetadataPType;

        let cfg = GoosefsConfig::new("localhost:0");
        let opts = super::list_status_p_options(
            Some(cfg.file_metadata_load_type),
            Some(cfg.file_metadata_sync_interval),
        );
        assert_eq!(
            opts.load_metadata_type,
            Some(LoadMetadataPType::Once as i32)
        );
        assert_eq!(
            opts.common_options
                .as_ref()
                .and_then(|c| c.sync_interval_ms),
            Some(-1)
        );
    }

    /// Java `commonDefaults(conf, true)` always sets `syncIntervalMs` and a
    /// non-empty `operationId`.
    #[test]
    fn write_common_p_options_matches_java_mutating_defaults() {
        let opts = super::write_common_p_options(-1);
        assert_eq!(opts.sync_interval_ms, Some(-1));
        let op = opts
            .operation_id
            .expect("mutating RPCs must send operationId");
        assert!(
            op.most_significant_bits.is_some() && op.least_significant_bits.is_some(),
            "FsOpPId must carry both UUID halves"
        );
    }

    /// Java `commonDefaults(conf)` (getStatus / listStatus / scheduleAsyncPersist)
    /// sets only `syncIntervalMs`.
    #[test]
    fn common_p_options_read_path_has_no_operation_id() {
        let opts = super::common_p_options(-1, None);
        assert_eq!(opts.sync_interval_ms, Some(-1));
        assert!(opts.operation_id.is_none());
    }

    /// Caller-supplied common_options win; only unset fields are filled.
    #[test]
    fn fill_write_common_options_preserves_caller_values() {
        let caller_id = super::new_fs_op_id();
        let mut slot = Some(super::common_p_options(0, Some(caller_id)));
        super::fill_write_common_options(&mut slot, -1);
        let filled = slot.expect("slot stays Some");
        assert_eq!(
            filled.sync_interval_ms,
            Some(0),
            "caller syncIntervalMs must not be overwritten"
        );
        assert_eq!(filled.operation_id, Some(caller_id));
    }

    #[test]
    fn fill_write_common_options_fills_empty_slot() {
        let mut slot = None;
        super::fill_write_common_options(&mut slot, -1);
        let filled = slot.expect("empty slot is filled");
        assert_eq!(filled.sync_interval_ms, Some(-1));
        assert!(filled.operation_id.is_some());
    }

    /// Java `renameDefaults` reads `goosefs.user.file.persist.on.rename`
    /// (default false). The config default and an explicit true must both
    /// reach `RenamePOptions.persist`.
    #[test]
    fn rename_p_options_persist_follows_config() {
        let cfg = GoosefsConfig::new("localhost:0");
        let opts =
            super::rename_p_options(cfg.file_metadata_sync_interval, cfg.file_persist_on_rename);
        assert_eq!(opts.persist, Some(false));
        assert_eq!(
            opts.common_options
                .as_ref()
                .and_then(|c| c.sync_interval_ms),
            Some(-1)
        );
        assert!(opts
            .common_options
            .as_ref()
            .and_then(|c| c.operation_id)
            .is_some());

        let opts = super::rename_p_options(-1, true);
        assert_eq!(opts.persist, Some(true));
    }
}