goosefs-sdk 0.2.1

// 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.

//! High-level file reader that orchestrates the complete read pipeline.
//!
//! `GoosefsFileReader` ties together all low-level components into a single
//! easy-to-use API, analogous to Java's `GoosefsFileInStream`:
//!
//! ```text
//! GoosefsFileReader::open_with_context(ctx, path)
//!   → MasterClient.get_status()          — get file metadata + block IDs
//!   → BlockMapper.plan_read()            — split file range → block segments
//!   → for each block segment:
//!       → WorkerRouterView.select_worker()  — consistent hash routing
//!       → WorkerClient.connect()         — connect to target worker (pooled)
//!       → GrpcBlockReader.open()         — open streaming read
//!       → reader.read_all()              — read all chunk data
//!   → concatenate results
//! ```
//!
//! # Example
//!
//! ```rust,no_run
//! use std::sync::Arc;
//! use goosefs_sdk::io::GoosefsFileReader;
//! use goosefs_sdk::context::FileSystemContext;
//! use goosefs_sdk::config::GoosefsConfig;
//!
//! # async fn example() -> goosefs_sdk::error::Result<()> {
//! let ctx = FileSystemContext::connect(GoosefsConfig::new("127.0.0.1:9200")).await?;
//!
//! // Read entire file
//! let data = GoosefsFileReader::read_file_with_context(ctx.clone(), "/my-file.txt").await?;
//! println!("read {} bytes", data.len());
//!
//! // Range read (offset=100, length=500)
//! let data = GoosefsFileReader::read_range_with_context(ctx.clone(), "/my-file.txt", 100, 500).await?;
//!
//! // Or use the builder for streaming reads
//! let mut reader = GoosefsFileReader::open_with_context(ctx.clone(), "/my-file.txt").await?;
//! while let Some(chunk) = reader.read_next_block().await? {
//!     println!("got {} bytes from block", chunk.len());
//! }
//! # Ok(())
//! # }
//! ```

use std::sync::Arc;

use bytes::{Bytes, BytesMut};
use tokio::sync::RwLock;
use tracing::{debug, warn};

use crate::block::mapper::{BlockMapper, BlockReadPlan};
use crate::block::router::WorkerRouterView;
use crate::cache::{
    page_cache_eligible, read_through_cache, CacheManager, ExternalRangeReader, FillMode,
};
use crate::client::worker::WorkerClientPool;
use crate::client::WorkerClient;
use crate::config::GoosefsConfig;
use crate::context::FileSystemContext;
use crate::error::{Error, Result};
use crate::fs::options::ufs_block_length;
use crate::io::reader::GrpcBlockReader;
use crate::proto::grpc::block::WorkerInfo;
use crate::proto::grpc::file::FileInfo;
use crate::proto::proto::dataserver::OpenUfsBlockOptions;

/// High-level file reader that orchestrates the full Goosefs read pipeline.
///
/// This struct encapsulates the complete read flow:
/// 1. `GetStatus` on Master to obtain file metadata (block IDs, block size, length)
/// 2. Discover workers and build consistent hash router
/// 3. Plan the read via `BlockMapper` (map file range → block segments)
/// 4. For each block: select worker → connect → stream-read via `GrpcBlockReader`
/// 5. Concatenate results
pub struct GoosefsFileReader {
    /// The Goosefs config.
    config: GoosefsConfig,
    /// The file path being read.
    path: String,
    /// File info from Master (contains block IDs, block size, length).
    ///
    /// Stored as `Arc<FileInfo>` for cheap sharing after reader initialization.
    /// Metadata-cache hits are first cloned into an owned `FileInfo` so
    /// CheckBlocks enrichment cannot mutate the cached Master snapshot.
    file_info: Arc<FileInfo>,
    /// Worker router for block → worker mapping.
    /// Worker router view for block → worker mapping.
    ///
    ///  Step 2
    /// migrated from `WorkerRouter` (per-reader `ArcSwap`×3) to
    /// `WorkerRouterView` (per-reader `Arc`×2 + `Option<i64>` value).
    /// Byte-exact routing behaviour is guaranteed by
    /// `block::router::tests::test_view_select_worker_matches_shared_for_all_block_ids`.
    router: WorkerRouterView,
    /// Optional shared worker-connection pool.
    ///
    /// Non-`None` when constructed via `*_with_context`: worker connections
    /// are acquired from the shared `WorkerClientPool` (zero TCP+SASL
    /// handshake on cache hit). `None` falls back to per-call
    /// `WorkerClient::connect` (legacy path).
    worker_pool: Option<Arc<WorkerClientPool>>,
    /// Optional shared context (non-`None` when created via `*_with_context`).
    /// Kept alive to prevent context GC while the reader is in use.
    _context: Option<Arc<FileSystemContext>>,
    /// Block-level read plans (populated on open).
    plans: Vec<BlockReadPlan>,
    /// Index of the next block to read.
    current_plan_index: usize,
    /// Total bytes read so far.
    total_bytes_read: u64,
    /// Requested read offset in the file.
    offset: u64,
    /// Requested read length.
    length: u64,

    // ── Local page cache plumbing (only populated by `*_with_context`) ───────
    /// Shared local page cache; `None` disables caching (falls back to the
    /// original worker-direct read, byte-for-byte unchanged). Mirrors the
    /// plumbing on [`crate::io::GoosefsFileInStream`] so both readers share
    /// the same on-disk pages.
    cache: Option<Arc<dyn CacheManager>>,
    /// Stable per-file cache namespace; derived identically to
    /// `GoosefsFileInStream` (`file_id.to_string()`, `unwrap_or(0)`).
    cache_file_id: Arc<str>,
    /// Effective page size (= `config.client_cache_page_size`).
    cache_page_size: u64,
    /// Whether missed pages are written back (opendal never `NoCache`, so this
    /// equals `cache.is_some()`).
    cache_fill: bool,
    /// Whether back-fill uses the bounded async write-back pool.
    cache_async_write: bool,

    // ── S4: pre-built UFS read options ───────────────────────────────────────
    /// Pre-built `OpenUfsBlockOptions` template for the UFS path, built once
    /// in [`Self::build`]. Cloned per segment with only `offset_in_file`
    /// updated — avoids re-cloning `ufs_path: String` + re-deriving
    /// `mount_id` / `no_cache` / `block_size` on every `read_segment` call.
    ///
    /// `None` when the file has no UFS path (cache-only data).
    ufs_read_options: Option<OpenUfsBlockOptions>,
}

// ── F1: on_file_open dedup cache ─────────────────────────────────────────
//
// `attach_cache` is called on every context open. When the same
// file is read repeatedly (e.g. Lance vector search hitting the same .lance
// file), `on_file_open` is invoked every time even though the (file_id, length,
// mtime) triple is identical. The `versions` RwLock read inside `on_file_open`
// is cheap when uncontended, but under 20+ concurrent queries the lock
// contention alone wastes ~5.5ms per call (result.log data).
//
// This process-level dedup cache records the last `on_file_open` check time
// per `(file_id, length, mtime)` key. If the same identity was checked within
// the TTL window, `on_file_open` is skipped entirely — zero lock acquisition.

/// TTL for the on_file_open dedup cache. Within this window, repeated opens of
/// the same file (same file_id + length + mtime) skip the `on_file_open` call
/// entirely. Set to 60s to match the `ConfigRefresher` cadence — if a file is
/// overwritten while the process is running, the next config refresh + open
/// will catch it. For safety, a server-detected mtime change always
/// re-triggers `on_file_open` (different key → miss).
const ON_FILE_OPEN_DEDUP_TTL: std::time::Duration = std::time::Duration::from_secs(60);

/// Key: (file_id as i64, length as i64, mtime as i64).
/// Value: Instant of the last `on_file_open` check.
type OnFileOpenDedupMap = std::collections::HashMap<(i64, i64, i64), std::time::Instant>;

static ON_FILE_OPEN_CACHE: std::sync::LazyLock<RwLock<OnFileOpenDedupMap>> =
    std::sync::LazyLock::new(|| RwLock::new(std::collections::HashMap::new()));

impl GoosefsFileReader {
    /// Open a file for reading using a shared [`FileSystemContext`].
    ///
    /// Reuses the Master client, worker-list snapshot and worker-connection
    /// pool cached inside the context — **no additional TCP+SASL handshake**
    /// to Master or Worker Manager is performed. This is the recommended
    /// constructor for long-running clients (OpenDAL, Lance, etc.).
    pub async fn open_with_context(ctx: Arc<FileSystemContext>, path: &str) -> Result<Self> {
        Self::open_inner(ctx, path, None, true).await
    }

    /// Open a file for range reading using a shared [`FileSystemContext`].
    ///
    /// Same benefits as [`open_with_context`](Self::open_with_context) plus
    /// explicit `(offset, length)` control.
    pub async fn open_range_with_context(
        ctx: Arc<FileSystemContext>,
        path: &str,
        offset: u64,
        length: u64,
    ) -> Result<Self> {
        Self::open_inner(ctx, path, Some((offset, length)), true).await
    }

    /// Shared opener for the context path.
    ///
    /// `range = None` reads the whole file. `use_page_cache` is `true` for the
    /// streaming constructors (`open_with_context` / `open_range_with_context`,
    /// OpenDAL / Lance) and `false` for the one-shot helpers (`read_file` /
    /// `read_range`), which stay worker-direct per the page-cache contract.
    async fn open_inner(
        ctx: Arc<FileSystemContext>,
        path: &str,
        range: Option<(u64, u64)>,
        use_page_cache: bool,
    ) -> Result<Self> {
        let (file_info, router) = Self::init_with_context(&ctx, path).await?;
        let (offset, length) = match range {
            Some((offset, length)) => (offset, length),
            None => (0, file_info.length.unwrap_or(0) as u64),
        };
        let config = ctx.config().clone();
        let pool = Some(ctx.acquire_worker_pool());
        let mut reader = Self::build(
            &config,
            path,
            file_info,
            router,
            pool,
            Some(ctx.clone()),
            offset,
            length,
        )?;
        reader.attach_cache(&ctx, use_page_cache).await;
        Ok(reader)
    }

    /// Internal: fetch file info via the shared Master, snapshot workers from
    /// the shared router — **no new RPC connections**.
    ///
    /// This is the context-aware analogue of [`Self::init`]. It mirrors the
    /// pattern used by `GoosefsFileWriter::create_with_context`: a local
    /// `WorkerRouterView` is created and seeded from the shared router's current
    /// snapshot, so per-read failure marking stays local and does not pollute
    /// the long-lived context-level routing state.
    ///
    ///: the local view is
    /// built via [`WorkerRouterView::from_shared`], which clones the shared
    /// router's `workers` + `hash_ring` `Arc`s wait-free (two `Arc::clone`s
    /// plus a value copy of `local_worker_id`) with **no `ArcSwap`
    /// allocation** — replacing the previous
    /// `WorkerRouter::snapshot_from` that also allocated three per-reader
    /// `ArcSwap` fields. `init_with_context`'s hot-path CPU drops from
    /// ~10.4 % to ~0 % of on-CPU, and per-reader Drop cost drops from
    /// ~19 % (`arc_swap::debt::list::LocalNode::with`) to ~0 %.
    /// Failure isolation is preserved: the view has its own
    /// `failed_workers` DashMap.
    async fn init_with_context(
        ctx: &Arc<FileSystemContext>,
        path: &str,
    ) -> Result<(Arc<FileInfo>, WorkerRouterView)> {
        // Shared Master + metadata cache (status / NotFound / incomplete fall-through).
        // CheckBlocks enrichment below mutates this owned copy (INV-MC-D1).
        let mut file_info = ctx.get_file_info_cached(path).await?;
        Self::reject_directory(&file_info, path)?;

        let file_length = file_info.length.unwrap_or(0);
        if file_length == 0 {
            debug!(path = %path, "file is empty");
        }

        debug!(
            path = %path,
            file_length = file_length,
            block_count = file_info.block_ids.len(),
            block_size = ?file_info.block_size_bytes,
            "fetched file metadata (via context)"
        );

        // 2. Snapshot from the shared router — clones two `Arc`s (workers +
        //    hash_ring) wait-free; does NOT rebuild the ring. The shared
        //    router is kept fresh by the context's background worker-refresh
        //    task, so no extra RPC is issued here.
        let shared_router = ctx.acquire_router();
        // Cheap non-empty guard (H4: use `workers_is_empty()` instead of
        // `get_workers().await.len()` — avoids an `Arc::clone` + `Drop`
        // just for a non-empty check).
        if shared_router.workers_is_empty() {
            return Err(Error::NoWorkerAvailable {
                message: "no workers available for reading".to_string(),
            });
        }
        let router = WorkerRouterView::from_shared(&shared_router);
        debug!("reusing worker snapshot from context (A1)");

        let pool = ctx.acquire_worker_pool();
        let config = ctx.config();
        crate::block::maybe_enrich_file_block_locations(
            &mut file_info,
            &router,
            Some(&pool),
            config,
            config.check_block_replicas,
        )
        .await;

        Ok((Arc::new(file_info), router))
    }

    /// Reject a directory the same way Java `BaseFileSystem.openFile` does.
    ///
    /// Directories report `length == 0`, so skipping this check makes
    /// `read_file` / `read_range` return empty bytes instead of
    /// [`Error::OpenDirectory`].
    fn reject_directory(file_info: &FileInfo, path: &str) -> Result<()> {
        if file_info.folder.unwrap_or(false) {
            return Err(Error::OpenDirectory {
                path: path.to_string(),
            });
        }
        Ok(())
    }

    /// Internal: build the reader from file info and router.
    #[allow(clippy::too_many_arguments)]
    fn build(
        config: &GoosefsConfig,
        path: &str,
        file_info: Arc<FileInfo>,
        router: WorkerRouterView,
        worker_pool: Option<Arc<WorkerClientPool>>,
        context: Option<Arc<FileSystemContext>>,
        offset: u64,
        length: u64,
    ) -> Result<Self> {
        // Plan the read
        let plans = BlockMapper::plan_read(&file_info, offset, length);
        debug!(
            path = %path,
            offset = offset,
            length = length,
            block_segments = plans.len(),
            "read plan created"
        );

        // S4: pre-build the UFS read options template once. Per-segment
        // `build_ufs_read_options` now clones this and updates only the two
        // block-geometry fields, avoiding a `ufs_path.clone()` + field
        // re-derivation on every `read_segment`.
        let ufs_read_options = {
            let ufs_path = file_info.ufs_path.as_ref();
            if ufs_path.map_or(true, |p| p.is_empty()) {
                None
            } else {
                Some(OpenUfsBlockOptions {
                    ufs_path: file_info.ufs_path.clone(),
                    // offset_in_file and block_size are per-segment; set to 0
                    // as placeholders, both overwritten by
                    // `build_ufs_read_options`.
                    offset_in_file: Some(0),
                    block_size: Some(0),
                    // Must be sent. The Worker admits a UFS block read only
                    // while `sessions_for(block) < maxUfsReadConcurrency`, and
                    // an absent field decodes as 0 — so once a block has any
                    // session entry, every later read of it waits on a permit
                    // that can never come. The first read of a block succeeds
                    // (no entry yet), which is why leaving this unset surfaced
                    // as "the *second* read of a path hangs forever".
                    // `GoosefsFileInStream` and the binding's `positioned_read`
                    // already send the Java default; match them.
                    max_ufs_read_concurrency: Some(
                        crate::fs::options::InStreamOptions::default().max_ufs_read_concurrency,
                    ),
                    mount_id: file_info.mount_id,
                    no_cache: Some(!file_info.cacheable.unwrap_or(true)),
                    user: None,
                    caller_type: None,
                    file_length: file_info.length,
                })
            }
        };

        Ok(Self {
            config: config.clone(),
            path: path.to_string(),
            file_info,
            router,
            worker_pool,
            _context: context,
            plans,
            current_plan_index: 0,
            total_bytes_read: 0,
            offset,
            length,
            // The cache is disabled by default here; `attach_cache()` enables
            // it in the async opener when a shared context is present.
            cache: None,
            cache_file_id: Arc::from(""),
            cache_page_size: config.client_cache_page_size,
            cache_fill: false,
            cache_async_write: false,
            ufs_read_options,
        })
    }

    /// Inject the local page cache from a shared [`FileSystemContext`]
    /// (best-effort).
    ///
    /// `use_page_cache` is `false` for the one-shot `read_file` / `read_range`
    /// helpers: those stay on the worker-direct path (Python binding contract
    /// and `docs/CLIENT_PAGE_CACHE_DESIGN.md` §14.1). Streaming
    /// `open_with_context` still attaches the cache for OpenDAL / Lance.
    ///
    /// `build()` is synchronous, but `on_file_open` is async, so cache
    /// activation is deferred to this async opener helper. The `file_id`,
    /// `length` and `mtime` derivations are kept **byte-for-byte aligned** with
    /// `URIStatus::from_proto` (all `unwrap_or(0)`), so this reader hits exactly
    /// the same on-disk pages as `GoosefsFileInStream`.
    async fn attach_cache(&mut self, ctx: &Arc<FileSystemContext>, use_page_cache: bool) {
        if !use_page_cache {
            self.cache = None;
            self.cache_fill = false;
            return;
        }

        // HR-1 (design ): `file_id <= 0` means no stable inode identity.
        // See [`page_cache_eligible`]. This guard MUST run before
        // `acquire_cache_manager()` / `on_file_open()` so a "0"-keyed open never
        // pollutes the version table.
        let file_id = self.file_info.file_id.unwrap_or(0);
        if !page_cache_eligible(file_id) {
            self.cache = None;
            self.cache_fill = false;
            return;
        }

        let cache = ctx.acquire_cache_manager();
        // Key on `file_id` (> 0), aligned with `URIStatus.file_id`. Do NOT fall
        // back to `path` — that would key differently from the InStream's "0"
        // and would go stale on same-name recreate.
        let cache_file_id: Arc<str> = Arc::from(file_id.to_string());

        // F1: Skip `on_file_open` if this exact (file_id, length, mtime) was
        // already checked within the dedup TTL window. The dedup cache is keyed
        // by the server-reported identity, so a genuine overwrite (different
        // length or mtime) always produces a different key → cache miss →
        // `on_file_open` runs and detects the change.
        if let Some(cache) = &cache {
            let length = self.file_info.length.unwrap_or(0);
            let mtime = self.file_info.last_modification_time_ms.unwrap_or(0);
            let dedup_key = (file_id, length, mtime);

            let skip = {
                let guard = ON_FILE_OPEN_CACHE.read().await;
                guard
                    .get(&dedup_key)
                    .is_some_and(|last| last.elapsed() < ON_FILE_OPEN_DEDUP_TTL)
            };
            if !skip {
                cache.on_file_open(&cache_file_id, length, mtime).await;
                let mut guard = ON_FILE_OPEN_CACHE.write().await;
                // Bounded: evict expired entries when the cache grows beyond a
                // reasonable size to prevent unbounded growth from many files.
                if guard.len() > 4096 {
                    guard.retain(|_, last| last.elapsed() < ON_FILE_OPEN_DEDUP_TTL);
                }
                guard.insert(dedup_key, std::time::Instant::now());
            }
        }

        let cfg = ctx.config();
        self.cache_fill = cache.is_some(); // opendal read path is never NoCache
        self.cache_async_write = cfg.client_cache_async_write_enabled;
        self.cache_page_size = cfg.client_cache_page_size;
        self.cache_file_id = cache_file_id;
        self.cache = cache;
    }

    /// Read the next block segment and return its data.
    ///
    /// Returns `None` when all block segments have been read.
    /// This is useful for streaming reads where you want to process
    /// data block-by-block.
    ///
    /// # Authentication failure recovery
    ///
    /// When a cached worker connection's SASL stream has expired, block reads
    /// fail with `AuthenticationFailed`.  This method detects such failures
    /// and automatically:
    /// 1. Invalidates the stale connection in the pool
    /// 2. Reconnects with fresh authentication
    /// 3. Retries the block read **once**
    ///
    /// This handles the common case of SASL expiry after process fork
    /// (Python `multiprocessing.spawn`) or long idle periods.
    pub async fn read_next_block(&mut self) -> Result<Option<Bytes>> {
        // 1) Skip invalid blocks and compute this segment's absolute file range.
        //    (Same invalid-block-skip semantics as the pre-cache implementation.)
        let (abs_offset, abs_end) = loop {
            if self.current_plan_index >= self.plans.len() {
                return Ok(None);
            }
            let plan = self.plans[self.current_plan_index].clone();
            let block_id = self.resolve_block_id(&plan);
            if block_id <= 0 {
                warn!(
                    block_index = plan.block_index,
                    plan_block_id = plan.block_id,
                    "invalid block ID, skipping block"
                );
                self.current_plan_index += 1;
                continue;
            }
            let block_size = self.file_info.block_size_bytes.unwrap_or(64 * 1024 * 1024) as u64;
            let abs_offset = (plan.block_index * block_size + plan.offset_in_block) as i64;
            break (abs_offset, abs_offset + plan.length as i64);
        };

        // 2) Fetch the bytes. With a cache, route through `read_through_cache`
        //    (this reader acts as the miss source via `ExternalRangeReader`,
        //    which uses `positioned_read` per page — the streaming ReadBlock
        //    verb deadlocks on the 2nd page of the same block). Otherwise read
        //    the whole segment directly (pre-cache worker-direct path).
        let data = match self.cache.clone() {
            Some(cache) => {
                let file_id = self.cache_file_id.clone();
                let page_size = self.cache_page_size;
                let file_length = self.file_length() as i64;
                let fill_mode = if !self.cache_fill {
                    FillMode::None
                } else if self.cache_async_write {
                    FillMode::Async
                } else {
                    FillMode::Sync
                };
                // The scalar plan values are already copied out above, so there
                // is no live borrow of `self` — we can pass `&mut self` as the
                // miss source to `read_through_cache`.
                read_through_cache(
                    &cache,
                    self,
                    &file_id,
                    page_size,
                    file_length,
                    abs_offset,
                    abs_end,
                    fill_mode,
                )
                .await?
            }
            None => self.read_file_range(abs_offset, abs_end, false).await?,
        };

        // 3) Advance the iterator state (identical to the old implementation).
        let bytes_read = data.len() as u64;
        self.total_bytes_read += bytes_read;
        self.current_plan_index += 1;

        debug!(
            abs_offset = abs_offset,
            abs_end = abs_end,
            bytes_read = bytes_read,
            total_read = self.total_bytes_read,
            cache_enabled = self.cache.is_some(),
            "block segment read complete"
        );

        Ok(Some(data))
    }

    /// Stateless: read a single block `plan`'s bytes (`[offset_in_block, +length)`).
    ///
    /// Preserves the full **two-layer failover** of the original
    /// `read_next_block`:
    ///   ① connect failure → auth failure reconnects; non-auth marks the worker
    ///      failed and retries once on a different worker;
    ///   ② auth failure during the RPC → single-flight reconnect + re-read
    ///      (same `open + read_all` verb, byte-equivalent — HR-3).
    ///
    /// Uses only `&self` — it never touches iterator state, so it is safe to be
    /// re-entered as the cache-miss source.
    ///
    /// `positioned = true` uses `GrpcBlockReader::positioned_read`
    /// (`position_short`), which serves exactly `length` bytes and ends the
    /// stream — the right shape for page-cache fills, which issue many small
    /// reads against the **same** block. `positioned = false` opens the
    /// streaming `ReadBlock` verb for the whole segment.
    async fn read_segment(
        &self,
        block_id: i64,
        plan: &BlockReadPlan,
        positioned: bool,
    ) -> Result<Bytes> {
        let ufs_options = self.build_ufs_read_options(plan);

        // ① Select worker + connection failover (mirrors the old read_next_block).
        // Prefer BlockInfo.locations (Java getInStream); hash when empty/unmatched.
        // Candidate width = max(maxRetryNode, replication) like Java getInStream.
        let locations = self.block_locations(block_id);
        let worker_info = self
            .router
            .select_worker_for_read(
                block_id,
                locations,
                self.config.file_replication_number,
                self.config.file_read_max_node_retry,
            )
            .await?;
        let worker_addr = Self::worker_addr(&worker_info)?;
        let worker = match self.acquire_worker(&worker_addr).await {
            Ok(w) => w,
            Err(e) if e.is_authentication_failed() => {
                debug!(
                    worker = %worker_addr,
                    error = %e,
                    "authentication failed on connect, reconnecting"
                );
                self.reconnect_worker(&worker_addr, None).await?
            }
            Err(e) => {
                // Non-auth connect failure: mark the worker failed and retry
                // once on a different worker.
                if let Some(a) = worker_info.address.as_ref() {
                    self.router.mark_failed(a);
                }
                warn!(
                    worker = %worker_addr,
                    error = %e,
                    "worker connection failed, trying another worker"
                );
                let retry = self
                    .router
                    .select_worker_for_read(
                        block_id,
                        self.block_locations(block_id),
                        self.config.file_replication_number,
                        self.config.file_read_max_node_retry,
                    )
                    .await
                    .map_err(|_| e)?;
                let retry_addr = Self::worker_addr(&retry)?;
                self.acquire_worker(&retry_addr).await?
            }
        };

        // ② Read the segment + single-flight auth reconnect on RPC failure.
        let worker_generation = worker.generation();
        match self
            .try_read_block(&worker, block_id, plan, ufs_options.clone(), positioned)
            .await
        {
            Ok(d) => Ok(d),
            Err(e) if e.is_authentication_failed() => {
                debug!(
                    block_id = block_id,
                    worker = %worker_addr,
                    stale_generation = worker_generation,
                    error = %e,
                    "auth failed during block read, requesting single-flight reconnect"
                );
                let fresh = self
                    .reconnect_worker(&worker_addr, Some(worker_generation))
                    .await?;
                self.try_read_block(&fresh, block_id, plan, ufs_options, positioned)
                    .await
            }
            Err(e) => Err(e),
        }
    }

    /// Stateless: read the absolute file range `[abs_offset, abs_end)`.
    ///
    /// Never touches `self.plans` / `self.current_plan_index` / `self.offset` /
    /// `self.length`, so it is safe to be re-entered as the cache-miss source
    /// (page-level back-fill) without corrupting the outer `read_next_block`
    /// iteration.
    async fn read_file_range(
        &self,
        abs_offset: i64,
        abs_end: i64,
        positioned: bool,
    ) -> Result<Bytes> {
        if abs_end <= abs_offset {
            return Ok(Bytes::new());
        }
        // Local re-planning: a page (≤ 1 MiB) usually lands inside a single
        // block (≥ 64 MiB), so this is typically one segment.
        let plans = BlockMapper::plan_read(
            &self.file_info,
            abs_offset as u64,
            (abs_end - abs_offset) as u64,
        );

        let mut buf = BytesMut::with_capacity((abs_end - abs_offset) as usize);
        for plan in &plans {
            let block_id = self.resolve_block_id(plan);
            // Same invalid-block-skip semantics as `read_next_block` (HR-1):
            // a non-positive `block_id` means the master returned no valid
            // block for this segment, so we skip it silently rather than
            // fail the whole range read. Any real hole in the file surfaces
            // via the `data.is_empty()` check below on a *valid* block.
            if block_id <= 0 {
                continue;
            }
            let data = self.read_segment(block_id, plan, positioned).await?;
            if data.is_empty() {
                return Err(Error::Internal {
                    message: format!("read_file_range: 0 bytes for block {block_id}"),
                    source: None,
                });
            }
            buf.extend_from_slice(&data);
        }
        Ok(buf.freeze())
    }

    /// Format a `WorkerInfo`'s address as `host:rpc_port`.
    ///
    /// Shared by the primary read and the failover retry so both build the
    /// address identically (inlines the old manual formatting + the
    /// `worker has no address` error branch).
    ///
    /// delegates to the shared [`rpc_endpoint`] helper (uses
    /// `itoa::write` for the port) so every `host:port` construction in
    /// the crate goes through the same allocation-free path.
    fn worker_addr(worker_info: &WorkerInfo) -> Result<String> {
        let addr = worker_info
            .address
            .as_ref()
            .ok_or_else(|| Error::Internal {
                message: "worker has no address".to_string(),
                source: None,
            })?;
        Ok(crate::block::router::rpc_endpoint(addr))
    }

    /// Read all remaining data and return it as a single `Bytes`.
    ///
    /// This reads all block segments sequentially and concatenates the results.
    pub async fn read_all(&mut self) -> Result<Bytes> {
        let expected_len = self.plans.iter().map(|p| p.length).sum::<u64>();
        let mut buf = BytesMut::with_capacity(expected_len as usize);

        while let Some(chunk) = self.read_next_block().await? {
            buf.extend_from_slice(&chunk);
        }

        Ok(buf.freeze())
    }

    /// Acquire a worker client for the given address.
    ///
    /// When a shared [`WorkerClientPool`] is available (context path), the
    /// connection is pulled from the pool — which caches authenticated gRPC
    /// channels per-address — so repeated reads to the same worker pay
    /// **zero** handshake cost.
    ///
    /// Without a pool (legacy config-only path), a fresh `WorkerClient` is
    /// established per call, matching the pre-context behaviour.
    async fn acquire_worker(&self, addr: &str) -> Result<WorkerClient> {
        if let Some(pool) = &self.worker_pool {
            pool.acquire(addr).await
        } else {
            WorkerClient::connect(addr, &self.config).await
        }
    }

    /// Invalidate the stale connection and reconnect with fresh authentication.
    ///
    /// Used when a block read or connect fails with `AuthenticationFailed`.
    ///
    /// When `stale_generation` is `Some(gen)`, the pool performs a
    /// **single-flight reconnect**: if another concurrent reader has already
    /// replaced the channel with a newer generation the existing fresh
    /// connection is returned without triggering a redundant TCP+SASL
    /// handshake.  When `None` (e.g. the failure happened during `connect`
    /// so no `WorkerClient` was ever produced) the pool performs an
    /// unconditional reconnect (still serialised per-address so concurrent
    /// callers share one handshake).
    async fn reconnect_worker(
        &self,
        addr: &str,
        stale_generation: Option<u64>,
    ) -> Result<WorkerClient> {
        if let Some(pool) = &self.worker_pool {
            match stale_generation {
                Some(gen) => pool.reconnect_if_stale(addr, gen).await,
                None => pool.reconnect(addr).await,
            }
        } else {
            WorkerClient::connect(addr, &self.config).await
        }
    }

    /// Attempt to open a block reader and read all data from it.
    ///
    /// Factored out from `read_next_block` so the auth-failure retry path
    /// can reuse the same logic with a fresh worker.
    ///
    /// `positioned` selects `position_short` (page-cache miss / random fill)
    /// vs the streaming `ReadBlock` verb (whole-segment sequential read).
    async fn try_read_block(
        &self,
        worker: &WorkerClient,
        block_id: i64,
        plan: &BlockReadPlan,
        ufs_options: Option<OpenUfsBlockOptions>,
        positioned: bool,
    ) -> Result<Bytes> {
        if positioned {
            return GrpcBlockReader::positioned_read(
                worker,
                block_id,
                plan.offset_in_block as i64,
                plan.length as i64,
                self.config.chunk_size as i64,
                ufs_options,
            )
            .await;
        }

        let mut block_reader = GrpcBlockReader::open(
            worker,
            block_id,
            plan.offset_in_block as i64,
            plan.length as i64,
            self.config.chunk_size as i64,
            ufs_options,
        )
        .await?;

        block_reader.read_all().await
    }

    /// Resolve the best block ID for a read plan.
    ///
    /// Prefers the block ID from `file_block_infos` (which contains the actual
    /// assigned block ID from the server) over the ID computed from `block_ids`.
    fn resolve_block_id(&self, plan: &BlockReadPlan) -> i64 {
        if let Some(fbi) = self
            .file_info
            .file_block_infos
            .get(plan.block_index as usize)
        {
            if let Some(bi) = &fbi.block_info {
                if let Some(id) = bi.block_id {
                    if id > 0 {
                        return id;
                    }
                }
            }
        }
        // Fall back to block_ids list
        plan.block_id
    }

    /// Master `BlockInfo.locations` for `block_id`, or empty when unavailable.
    fn block_locations(&self, block_id: i64) -> &[crate::proto::grpc::BlockLocation] {
        self.file_info
            .file_block_infos
            .iter()
            .find_map(|fbi| {
                let bi = fbi.block_info.as_ref()?;
                if bi.block_id == Some(block_id) {
                    Some(bi.locations.as_slice())
                } else {
                    None
                }
            })
            .unwrap_or(&[])
    }

    /// Build `OpenUfsBlockOptions` for a block that may reside in UFS.
    ///
    /// When data was written with `THROUGH` mode, the block only exists in the
    /// underlying file system (UFS). The Worker needs `OpenUfsBlockOptions` to
    /// know the UFS path, mount ID, and block geometry so it can read the data
    /// from UFS on behalf of the client.
    ///
    /// Returns `None` if the file has no UFS path (i.e. data is cache-only).
    ///
    /// clones the pre-built `self.ufs_read_options` template and updates only
    /// the block-geometry fields. The old path re-cloned `ufs_path: String` +
    /// re-derived `mount_id` / `no_cache` on every `read_segment` call. Now the
    /// per-segment cost is one `OpenUfsBlockOptions::clone` (which clones
    /// `ufs_path: Option<String>` — still a String clone, but only one field
    /// instead of re-deriving all fields) plus two assignments. The `ufs_path`
    /// clone is unavoidable because the proto type owns its string.
    fn build_ufs_read_options(&self, plan: &BlockReadPlan) -> Option<OpenUfsBlockOptions> {
        let template = self.ufs_read_options.as_ref()?;
        let nominal_block_size = self.file_info.block_size_bytes.unwrap_or(64 * 1024 * 1024);
        let offset_in_file = (plan.block_index as i64).saturating_mul(nominal_block_size);

        let mut opts = template.clone();
        opts.offset_in_file = Some(offset_in_file);
        // The real length of *this* block, not the nominal block size — see
        // [`ufs_block_length`]. A `FileInfo` without a length cannot be
        // clamped against anything, so fall back to the nominal size.
        opts.block_size = Some(match self.file_info.length {
            Some(len) => ufs_block_length(len, nominal_block_size, plan.block_index),
            None => nominal_block_size,
        });
        Some(opts)
    }

    /// One-shot convenience: read an entire file using a shared context.
    ///
    /// Reuses Master and Worker connections from the supplied
    /// [`FileSystemContext`]. This is the recommended entry point for
    /// long-running clients that want to avoid per-call handshakes.
    ///
    /// Returns [`Error::OpenDirectory`] if `path` is a directory.
    ///
    /// ```rust,no_run
    /// # async fn example() -> goosefs_sdk::error::Result<()> {
    /// use std::sync::Arc;
    /// use goosefs_sdk::io::GoosefsFileReader;
    /// use goosefs_sdk::config::GoosefsConfig;
    /// use goosefs_sdk::context::FileSystemContext;
    ///
    /// let config = GoosefsConfig::new("127.0.0.1:9200");
    /// let ctx = FileSystemContext::connect(config).await?;
    /// let data = GoosefsFileReader::read_file_with_context(ctx, "/my-file.txt").await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn read_file_with_context(ctx: Arc<FileSystemContext>, path: &str) -> Result<Bytes> {
        // One-shot helpers stay worker-direct: attaching the page cache here
        // turns a single streamed block read into per-page positioned fills
        // and (before the miss source used `position_short`) hung on the 2nd
        // page. Streaming `open_with_context` is the cache-aware path.
        let mut reader = Self::open_inner(ctx, path, None, false).await?;
        reader.read_all().await
    }

    /// One-shot convenience: read a byte range from a file using a shared context.
    pub async fn read_range_with_context(
        ctx: Arc<FileSystemContext>,
        path: &str,
        offset: u64,
        length: u64,
    ) -> Result<Bytes> {
        let mut reader = Self::open_inner(ctx, path, Some((offset, length)), false).await?;
        reader.read_all().await
    }

    /// Read multiple `(offset, length)` ranges from `path` in one call
    ///
    ///
    /// # Behaviour
    ///
    /// The returned `Vec<Bytes>` has **exactly the same length and order**
    /// as `ranges`, and `result[i]` is byte-identical to what
    /// [`Self::read_range_with_context`] would return for
    /// `ranges[i]` — regardless of whether coalescing is enabled.
    ///
    /// - `ctx.config().range_coalesce_enabled == false` (**default**): each
    ///   input range is served by an independent
    ///   `read_range_with_context` call. Behaviour is exactly what a
    ///   caller-side loop would produce, so this path is a drop-in
    ///   replacement.
    /// - `ctx.config().range_coalesce_enabled == true`: adjacent input
    ///   ranges (gap ≤ `range_coalesce_gap_bytes`) are merged into one
    ///   fetch, capped at `range_coalesce_max_bytes` per merged fetch,
    ///   and the payload is spliced back so each `result[i]` recovers
    ///   the exact bytes of `ranges[i]`. This trades over-read of
    ///   `≤ Σ gap_i` bytes for a large drop in H2 stream count, which
    ///   is what the flame graph identifies as the dominant cost on
    ///   Lance / DuckDB scan patterns.
    ///
    /// Empty input ranges (`len == 0`) produce empty `Bytes` in the
    /// output without triggering any I/O.
    ///
    /// # Errors
    ///
    /// If any underlying `read_range` call fails, that error is returned
    /// immediately. In the enabled path a merged fetch failure fails
    /// **all** its constituent input ranges (they share transport), which
    /// matches the failure model the H2 layer would produce anyway.
    pub async fn read_ranges_with_context(
        ctx: Arc<FileSystemContext>,
        path: &str,
        ranges: &[(u64, u64)],
    ) -> Result<Vec<Bytes>> {
        let cfg = ctx.config();
        // Fast path: feature off → verbatim per-range reads. This keeps
        // behaviour bit-identical to the pre-B2 baseline whenever the
        // opt-in flag is not set.
        if !cfg.range_coalesce_enabled {
            let mut out: Vec<Bytes> = Vec::with_capacity(ranges.len());
            for &(off, len) in ranges {
                if len == 0 {
                    out.push(Bytes::new());
                    continue;
                }
                let bytes = Self::read_range_with_context(ctx.clone(), path, off, len).await?;
                out.push(bytes);
            }
            return Ok(out);
        }

        // Enabled path: plan → issue merged fetches → splice back.
        let plan = crate::io::range_coalesce::plan(
            ranges,
            cfg.range_coalesce_gap_bytes,
            cfg.range_coalesce_max_bytes,
        );
        debug!(
            path = %path,
            input_count = ranges.len(),
            output_count = plan.fetches.len(),
            input_bytes = plan.total_input_bytes,
            fetch_bytes = plan.total_fetch_bytes,
            wasted_bytes = plan.wasted_bytes(),
            "range coalesce plan"
        );

        // Issue one `read_range` per merged fetch. We do them sequentially
        // to preserve error-order semantics (first failing fetch wins);
        // downstream callers already treat the whole batch as atomic.
        let mut fetch_bufs: Vec<Bytes> = Vec::with_capacity(plan.fetches.len());
        for f in &plan.fetches {
            let bytes = Self::read_range_with_context(ctx.clone(), path, f.offset, f.len).await?;
            // Defensive: the reader should return exactly `f.len` bytes.
            // If a short read slips through, fail loudly — the byte-
            // equivalence contract of `read_ranges` cannot be honoured.
            if bytes.len() as u64 != f.len {
                return Err(Error::BlockIoError {
                    message: format!(
                        "read_ranges: merged fetch (offset={}, len={}) returned {} bytes",
                        f.offset,
                        f.len,
                        bytes.len()
                    ),
                });
            }
            fetch_bufs.push(bytes);
        }

        // Splice each caller-visible slice out of its assigned fetch.
        // `Bytes::slice` is O(1) — it just clones the ref-count and
        // narrows the view — so this loop allocates nothing.
        Ok(Self::splice_from_plan(&plan, &fetch_bufs))
    }

    /// Pure splice layer for [`Self::read_ranges_with_context`]:
    /// given the coalesce plan and the concrete `Bytes` returned for
    /// each merged fetch, produce one `Bytes` per input range in the
    /// caller's original order.
    ///
    /// Extracted so this behaviour is trivially unit-testable offline
    /// (no `FileSystemContext` / master required). Every splice is an
    /// `O(1)` `Bytes::slice` — no allocation, no copy.
    ///
    /// # Preconditions
    ///
    /// - `fetch_bufs.len() == plan.fetches.len()`
    /// - `fetch_bufs[i].len() as u64 == plan.fetches[i].len` for all `i`
    ///
    /// The caller (`read_ranges_with_context`) enforces both by
    /// construction; violations panic here (they would indicate a
    /// transport-layer contract violation).
    fn splice_from_plan(
        plan: &crate::io::range_coalesce::CoalescePlan,
        fetch_bufs: &[Bytes],
    ) -> Vec<Bytes> {
        debug_assert_eq!(fetch_bufs.len(), plan.fetches.len());
        let mut out: Vec<Bytes> = Vec::with_capacity(plan.slices.len());
        for s in &plan.slices {
            if s.fetch_index == crate::io::range_coalesce::NO_FETCH {
                out.push(Bytes::new());
                continue;
            }
            let src = &fetch_bufs[s.fetch_index];
            let end = s.offset_in_fetch + s.len;
            out.push(src.slice(s.offset_in_fetch..end));
        }
        out
    }

    // ── Accessors ────────────────────────────────────────────────

    /// Get the file path being read.
    pub fn path(&self) -> &str {
        &self.path
    }

    /// Get the file info.
    pub fn file_info(&self) -> &FileInfo {
        &self.file_info
    }

    /// Get the total file length (from metadata).
    pub fn file_length(&self) -> u64 {
        self.file_info.length.unwrap_or(0) as u64
    }

    /// Get the total bytes read so far.
    pub fn bytes_read(&self) -> u64 {
        self.total_bytes_read
    }

    /// Get the number of block segments in the read plan.
    pub fn block_count(&self) -> usize {
        self.plans.len()
    }

    /// Get the index of the next block to be read.
    pub fn current_block_index(&self) -> usize {
        self.current_plan_index
    }

    /// Whether all blocks have been read.
    pub fn is_complete(&self) -> bool {
        self.current_plan_index >= self.plans.len()
    }

    /// Get the requested read offset.
    pub fn offset(&self) -> u64 {
        self.offset
    }

    /// Get the requested read length.
    pub fn length(&self) -> u64 {
        self.length
    }
}

/// Bridges the reader's worker/UFS range read to the cache layer's miss
/// source, so [`crate::cache::read_through_cache`] can drive page fills
/// from within [`GoosefsFileReader::read_next_block`].
///
/// Misses use `positioned_read` (`position_short`), one short-lived stream per
/// page. `read_file_range` is stateless (`&self`), so re-entry here does not
/// perturb `read_next_block` iteration state.
#[async_trait::async_trait]
impl ExternalRangeReader for GoosefsFileReader {
    async fn read_range(&mut self, offset: i64, end: i64) -> Result<Bytes> {
        self.read_file_range(offset, end, true).await
    }
}

// ── Unit tests (pure logic — no I/O) ─────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::fs::options::InStreamOptions;
    use crate::proto::grpc::WorkerNetAddress;

    /// Build a cache-disabled reader over a synthetic file for pure-logic tests.
    fn make_reader(length: i64, block_size: i64) -> GoosefsFileReader {
        let num_blocks = if block_size > 0 {
            (length + block_size - 1) / block_size
        } else {
            0
        };
        let block_ids: Vec<i64> = (1001..(1001 + num_blocks.max(0))).collect();
        let file_info = FileInfo {
            length: Some(length),
            block_size_bytes: Some(block_size),
            block_ids,
            completed: Some(true),
            folder: Some(false),
            ufs_path: Some(String::new()),
            ..Default::default()
        };
        let config = GoosefsConfig::new("127.0.0.1:9200");
        GoosefsFileReader::build(
            &config,
            "/synthetic",
            Arc::new(file_info),
            WorkerRouterView::empty(),
            None,
            None,
            0,
            length as u64,
        )
        .expect("build reader")
    }

    /// **Regression**: `OpenUfsBlockOptions.max_ufs_read_concurrency` must be
    /// sent on the one-shot read path.
    ///
    /// The Worker admits a UFS block read only while the block's session count
    /// is below this limit, and an absent `optional int32` decodes as `0`.
    /// Leaving it unset therefore let the first read of a block through and
    /// wedged every read after it — the "second `read_file` of a path hangs
    /// forever" report. `GoosefsFileInStream` and the Python binding's
    /// `positioned_read` always sent the Java default of 8; this path did not.
    #[test]
    fn ufs_read_options_carry_max_read_concurrency() {
        let bs = 64 * 1024 * 1024i64;
        let file_info = FileInfo {
            length: Some(bs / 2),
            block_size_bytes: Some(bs),
            block_ids: vec![1001],
            completed: Some(true),
            folder: Some(false),
            ufs_path: Some("cosn://bucket/blob.bin".to_string()),
            mount_id: Some(7),
            ..Default::default()
        };
        let reader = GoosefsFileReader::build(
            &GoosefsConfig::new("127.0.0.1:9200"),
            "/blob.bin",
            Arc::new(file_info),
            WorkerRouterView::empty(),
            None,
            None,
            0,
            (bs / 2) as u64,
        )
        .expect("build reader");

        let opts = reader
            .build_ufs_read_options(&reader.plans[0])
            .expect("a file with a ufs_path must produce OpenUfsBlockOptions");
        assert_eq!(
            opts.max_ufs_read_concurrency,
            Some(InStreamOptions::default().max_ufs_read_concurrency),
            "unset (or 0) makes the Worker refuse every read after the first"
        );
        assert!(opts.max_ufs_read_concurrency.unwrap() > 0);
    }

    /// **Regression**: `OpenUfsBlockOptions.block_size` must carry the real
    /// length of the block, not the file's nominal block size.
    ///
    /// The Worker treats it as the block length and asks the UFS for that many
    /// bytes when back-filling the page cache, so over-reporting a partial tail
    /// block makes the read come up short and the block is never cached:
    ///
    /// ```text
    /// ERROR LocalCacheManager - Failed to read page
    ///   BlockPageId{FileId=paged_block_503316480_size_1048576, PageIndex=0}:
    ///   supposed to read 1048576 bytes, 13 bytes actually read
    /// WARN  PagedBlockStore - Failed to cache block 503316480 in page mode
    /// ```
    ///
    /// Java sends `Math.min(length - blockSize * seq, blockSize)`.
    #[test]
    fn ufs_read_options_carry_actual_tail_block_length() {
        let bs = 1 << 20i64;
        let length = 2 * bs + 100;
        let file_info = FileInfo {
            length: Some(length),
            block_size_bytes: Some(bs),
            block_ids: vec![1001, 1002, 1003],
            completed: Some(true),
            folder: Some(false),
            ufs_path: Some("cosn://bucket/tail.lance".to_string()),
            mount_id: Some(7),
            ..Default::default()
        };
        let reader = GoosefsFileReader::build(
            &GoosefsConfig::new("127.0.0.1:9200"),
            "/tail.lance",
            Arc::new(file_info),
            WorkerRouterView::empty(),
            None,
            None,
            0,
            length as u64,
        )
        .expect("build reader");

        let seen: Vec<(i64, i64)> = reader
            .plans
            .iter()
            .map(|plan| {
                let opts = reader
                    .build_ufs_read_options(plan)
                    .expect("a file with a ufs_path must produce OpenUfsBlockOptions");
                (opts.offset_in_file.unwrap(), opts.block_size.unwrap())
            })
            .collect();

        assert_eq!(
            seen,
            vec![(0, bs), (bs, bs), (2 * bs, 100)],
            "the tail block must advertise 100 bytes, not the nominal {bs}"
        );
    }

    /// A file smaller than one block has no full block at all — block 0 is the
    /// tail. This is the shape of the Lance manifest / version-hint files that
    /// surfaced the bug.
    #[test]
    fn ufs_read_options_sub_block_file_reports_file_length() {
        let bs = 1 << 20i64;
        let file_info = FileInfo {
            length: Some(13),
            block_size_bytes: Some(bs),
            block_ids: vec![1001],
            completed: Some(true),
            folder: Some(false),
            ufs_path: Some("cosn://bucket/latest_version_hint.json".to_string()),
            mount_id: Some(7),
            ..Default::default()
        };
        let reader = GoosefsFileReader::build(
            &GoosefsConfig::new("127.0.0.1:9200"),
            "/latest_version_hint.json",
            Arc::new(file_info),
            WorkerRouterView::empty(),
            None,
            None,
            0,
            13,
        )
        .expect("build reader");

        let opts = reader
            .build_ufs_read_options(&reader.plans[0])
            .expect("a file with a ufs_path must produce OpenUfsBlockOptions");
        assert_eq!(opts.block_size, Some(13));
        assert_eq!(opts.offset_in_file, Some(0));
    }

    /// `read_file` / `read_range` must reject directories (Java
    /// `openFile` / `OpenDirectoryException`). Without this, a directory's
    /// length-0 metadata would return empty bytes.
    #[test]
    fn test_reject_directory() {
        let mut info = FileInfo {
            folder: Some(true),
            length: Some(0),
            ..Default::default()
        };
        let err = GoosefsFileReader::reject_directory(&info, "/d").unwrap_err();
        assert!(
            matches!(err, Error::OpenDirectory { ref path } if path == "/d"),
            "expected OpenDirectory, got {err:?}"
        );

        info.folder = Some(false);
        assert!(GoosefsFileReader::reject_directory(&info, "/f").is_ok());

        info.folder = None;
        assert!(GoosefsFileReader::reject_directory(&info, "/f").is_ok());
    }

    /// A freshly-built reader (no context) has caching off, so its read path is
    /// byte-for-byte the legacy worker-direct path.
    #[test]
    fn test_build_defaults_disable_cache() {
        let reader = make_reader(1024, 1024);
        assert!(reader.cache.is_none(), "cache must default to disabled");
        assert!(!reader.cache_fill, "fill must default to false");
    }

    /// HR-1: missing / non-positive `file_id` must keep the page cache off.
    /// (`attach_cache` uses [`page_cache_eligible`]; synthetic readers start
    /// with `file_id = None` → treated as 0.)
    #[test]
    fn hr1_non_positive_file_id_is_not_page_cache_eligible() {
        use crate::cache::page_cache_eligible;

        let reader = make_reader(1024, 1024);
        // Synthetic FileInfo defaults leave file_id unset → treated as 0.
        assert_eq!(reader.file_info.file_id, None);
        assert!(!page_cache_eligible(reader.file_info.file_id.unwrap_or(0)));
        assert!(reader.cache.is_none());

        assert!(!page_cache_eligible(0));
        assert!(!page_cache_eligible(-7));
        assert!(page_cache_eligible(42));
    }

    /// `worker_addr` formats `host:rpc_port` and defaults sensibly.
    #[test]
    fn test_worker_addr_formatting() {
        let wi = WorkerInfo {
            address: Some(WorkerNetAddress {
                host: Some("10.0.0.5".to_string()),
                rpc_port: Some(9207),
                ..Default::default()
            }),
            ..Default::default()
        };
        assert_eq!(
            GoosefsFileReader::worker_addr(&wi).unwrap(),
            "10.0.0.5:9207"
        );

        // Missing host/port fall back to the documented defaults.
        let wi_defaults = WorkerInfo {
            address: Some(WorkerNetAddress::default()),
            ..Default::default()
        };
        assert_eq!(
            GoosefsFileReader::worker_addr(&wi_defaults).unwrap(),
            "127.0.0.1:9203"
        );
    }

    /// A `WorkerInfo` with no address surfaces an internal error rather than
    /// panicking.
    #[test]
    fn test_worker_addr_missing_address_errors() {
        let wi = WorkerInfo {
            address: None,
            ..Default::default()
        };
        assert!(GoosefsFileReader::worker_addr(&wi).is_err());
    }

    // ── B2: read_ranges splice layer ──────────────────────────────

    /// The splice layer must return one `Bytes` per input range, in the
    /// caller's original order, byte-identical to a hypothetical
    /// standalone `read_range` for that input.
    #[test]
    fn splice_from_plan_reconstructs_caller_order_and_bytes() {
        use crate::io::range_coalesce::plan;

        // Inputs deliberately out of order to exercise sort + reorder.
        let inputs = &[(200u64, 10u64), (0, 20), (10, 5), (100, 4)];
        // gap = 100 → merges (0,20) and (10,5) into [0,25); (100,4) sits
        // alone (gap to nearest = 75 ≤ 100 → also merges into [0, 104]);
        // (200,10) is 96 bytes past the end → also merges. Use a small
        // cap to force a split so we cover both single- and
        // multi-fetch splice paths.
        let p = plan(inputs, 100, 60);
        assert!(!p.fetches.is_empty());

        // Simulate storage: byte(off) = (off & 0xff) as u8.
        let synth = |off: u64, len: u64| -> Bytes {
            let v: Vec<u8> = (0..len).map(|i| ((off + i) & 0xff) as u8).collect();
            Bytes::from(v)
        };
        let fetch_bufs: Vec<Bytes> = p.fetches.iter().map(|f| synth(f.offset, f.len)).collect();

        let out = GoosefsFileReader::splice_from_plan(&p, &fetch_bufs);
        assert_eq!(out.len(), inputs.len(), "output length must match input");
        for (i, &(off, len)) in inputs.iter().enumerate() {
            let expected = synth(off, len);
            assert_eq!(
                out[i].as_ref(),
                expected.as_ref(),
                "output[{i}] byte mismatch for input ({off},{len})"
            );
        }
    }

    /// Empty input ranges must produce empty `Bytes` without triggering
    /// a fetch, and must NOT shift the fetch indices of the real ranges.
    #[test]
    fn splice_from_plan_preserves_empty_ranges_at_original_position() {
        use crate::io::range_coalesce::plan;

        // (100, 0) is empty, (0, 10) is real, (7, 0) is empty, (50, 5) real.
        let inputs = &[(100u64, 0u64), (0, 10), (7, 0), (50, 5)];
        let p = plan(inputs, 0, 4096);
        // Two real ranges → two disjoint fetches (gap=0 disallows merge).
        assert_eq!(p.fetches.len(), 2);

        // Synth: byte(off) = (off & 0x7f)
        let synth = |off: u64, len: u64| -> Bytes {
            Bytes::from(
                (0..len)
                    .map(|i| ((off + i) & 0x7f) as u8)
                    .collect::<Vec<_>>(),
            )
        };
        let fetch_bufs: Vec<Bytes> = p.fetches.iter().map(|f| synth(f.offset, f.len)).collect();

        let out = GoosefsFileReader::splice_from_plan(&p, &fetch_bufs);
        assert_eq!(out.len(), 4);
        assert!(out[0].is_empty(), "empty input must yield empty Bytes");
        assert_eq!(out[1].len(), 10);
        assert!(out[2].is_empty(), "empty input must yield empty Bytes");
        assert_eq!(out[3].len(), 5);
    }

    /// Config default must keep `range_coalesce_enabled = false`, i.e.
    /// the `read_ranges` fast path serves each input verbatim
    /// (behaviour identical to a caller-side loop). This test guards
    /// against an accidental default flip.
    #[test]
    fn range_coalesce_disabled_by_default() {
        let cfg = GoosefsConfig::default();
        assert!(
            !cfg.range_coalesce_enabled,
            "range_coalesce_enabled must default to false"
        );
        assert_eq!(cfg.range_coalesce_gap_bytes, 64 * 1024);
        assert_eq!(cfg.range_coalesce_max_bytes, 4 * 1024 * 1024);
    }
}