From b58d52ac1683c5f77d88e2d959007c37b45c8594 Mon Sep 17 00:00:00 2001
From: Chris Lu <chrislusf@users.noreply.github.com>
Date: Tue, 25 Aug 2026 15:21:37 -0700
Subject: [PATCH] rust volume: search .sdx for read-only volumes instead of
 holding the index (#10951)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* rust volume: search .sdx for read-only volumes instead of holding the index

The Go volume server loads every read-only volume through SortedFileNeedleMap:
the index lives on disk as a sorted .sdx, a lookup is a binary search, and
since #10950 no descriptor is held between lookups. The Rust server had no
counterpart. Read-only volumes built a full in-memory CompactNeedleMap, and
cloud-tiered ones — noWriteCanDelete, so not the read-only branch — went
through the writable path and pinned an .idx append handle on top of it. At the
hundreds of thousands of tiered volumes a real server carries, that is an index
in RAM and a descriptor each, for volumes nobody reads.

Port the sorted map and the bounded handle pool. A tiered volume now costs zero
descriptors and zero index bytes when idle; the pool keeps the hot handles open
so a busy volume does not pay an open() per needle. Handles are Arc<File>, so
an eviction cannot close one a reader still holds.

The generated .sdx is byte-identical to Go's — same sort, same last-write-wins,
same dropped tombstones — so a volume moved between a Go and a Rust server reads
whichever copy is already on disk. A test pins the bytes against a Go-generated
fixture.

* rust volume: fail compaction on an unreadable .sdx, and rebuild the map on tier-down

Two ways the sorted map could lose data.

iter_entries swallowed read errors and returned however many entries it managed
to collect. Compaction takes that vector for the complete live set, so a
truncated .sdx or a mid-scan I/O fault would commit a volume missing every
needle past the failure. Return a Result instead and abort. redb's
collect_entries dropped errors the same way on the same path, so it goes with
it.

Tier-down clears the remote mode and publishes the volume as writable, but the
map it booted with is the read-only sorted one. Its put always fails, so the
first write would append to the local .dat and then fail to index it, leaving
bytes nothing references — and a non-fsync write repeats it. Fold the
reopen_idx_for_write swap into refresh_remote_write_mode so the map always
matches the mode it just published; a rebuild that fails pins the volume
read-only rather than letting it take writes it cannot record.

Go reaches neither: its tier-down leaves noWriteCanDelete set, so the volume
stays read-only until a reload or an explicit mark-writable, which already goes
through reopenIdxForWrite.

* rust volume: keep read-only volumes mountable on a read-only index dir, and batch the .sdx scan

Building .sdx writes to the index directory, and load_index_sorted_file also
created a missing .idx there. A volume whose index sits on a read-only mount
took both paths and failed to load, where before it mounted read-only off an
in-memory index and served reads. Create the .idx only where deletes are
allowed, and fall back to the in-memory map when the sorted one cannot be
built, so a directory nobody can write costs memory rather than availability.

The end-to-end scan behind iter_entries, ascending_visit and save_to_idx read
one entry per syscall. Read 1024 at a time instead, the batch size
idx::walk_index_file uses. Positional reads, not a cursor: the handle is shared
with any other borrower.

Also gate the Go byte-parity fixture on the 5bytes feature it describes, which
is otherwise dead code in a 4-byte-offset build.

* rust volume: roll back a failed writable mark, and rebuild a torn .sdx

set_writable clears the read-only flags before it can know the rest will
succeed, but only the map rebuild rolled them back. An .idx writer that fails to
attach left the volume advertising writable over a needle map with no writer, so
puts landed in memory and were gone after a restart — the exact failure the
function exists to prevent. The read-only-mount fallback made it reachable: that
path loads an in-memory map with no writer attached. All three steps now run
behind one rollback point.

A .sdx whose length is not a whole number of entries was accepted as long as it
looked fresh, and truncation is what makes it look fresh. The entry count then
floored, hiding the last needle from lookups and from compaction, which would
commit the shorter set. Treat a torn file like a stale one and rebuild it from
.idx. Go writes .sdx in place rather than through a temporary, so a crash
mid-generation is a real way to produce one.

Appends now start at the last whole .idx entry too, so a torn tail there is
overwritten by the next tombstone instead of misaligning every row after it.

* rust volume: trim a torn .idx before writing to it, keep delete-only volumes online, count sorted-map deletes

Three from review.

Flooring the sorted map's append offset only protected its own positional
writes. Every writable path appends at EOF instead, so a partial row left by a
short write pushed the next row off alignment and the following load parsed the
rest of the file as garbage. Drop the partial row before attaching any writable
index writer — it is unrecoverable anyway, and every loader already skips it.
Go refuses to load such a volume at all; trimming keeps it mountable with the
rows before the tear intact.

The unwritable-index-dir fallback stopped one step short for volumes that allow
deletes, which is every tiered one: the in-memory loader opens .idx read-write
there and fails on the same directory that just refused the .sdx, so the volume
stayed offline. Give up the deletes instead — without a writer no tombstone
could be recorded anyway — and a remount on a writable directory restores them.

Sorted-map deletes left the counters untouched, so a tiered volume reported
itself garbage-free until it restarted. They now land where a reload would put
them: the tombstone is another .idx row, and both it and the row it supersedes
count as deletions under the rule the load-time metric applies. Go skips this
too, and should not.
---
 seaweed-volume/src/server/grpc_server.rs      |  17 +-
 seaweed-volume/src/server/heartbeat.rs        |   4 +-
 seaweed-volume/src/storage/needle_map.rs      |  81 +-
 .../src/storage/needle_map/file_pool.rs       | 247 +++++
 .../src/storage/needle_map/sorted_file.rs     | 908 ++++++++++++++++++
 seaweed-volume/src/storage/volume.rs          | 753 ++++++++++++++-
 6 files changed, 1967 insertions(+), 43 deletions(-)
 create mode 100644 seaweed-volume/src/storage/needle_map/file_pool.rs
 create mode 100644 seaweed-volume/src/storage/needle_map/sorted_file.rs

diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs
index 3c82b680e..5d77e2818 100644
--- a/seaweed-volume/src/server/grpc_server.rs
+++ b/seaweed-volume/src/server/grpc_server.rs
@@ -3575,7 +3575,12 @@ impl VolumeServer for VolumeGrpcService {
                             modified_time: dat_modified_secs,
                             extension: ".dat".to_string(),
                         });
-                        vol.refresh_remote_write_mode();
+                        vol.refresh_remote_write_mode().map_err(|e| {
+                            Status::internal(format!(
+                                "volume {} failed to refresh write mode: {}",
+                                vid, e
+                            ))
+                        })?;
 
                         if let Err(e) = vol.save_volume_info() {
                             return Err(Status::internal(format!(
@@ -3764,7 +3769,15 @@ impl VolumeServer for VolumeGrpcService {
                     if !vol.volume_info.files.is_empty() {
                         vol.volume_info.files.remove(0);
                     }
-                    vol.refresh_remote_write_mode();
+                    // Swaps the read-only sorted map out before the volume is
+                    // published as writable; without it the first write would
+                    // append to the local .dat and then fail to index.
+                    vol.refresh_remote_write_mode().map_err(|e| {
+                        Status::internal(format!(
+                            "volume {} failed to refresh write mode: {}",
+                            vid, e
+                        ))
+                    })?;
 
                     if let Err(e) = vol.save_volume_info() {
                         return Err(Status::internal(format!(
diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs
index 723056f65..f0a23c314 100644
--- a/seaweed-volume/src/server/heartbeat.rs
+++ b/seaweed-volume/src/server/heartbeat.rs
@@ -1578,7 +1578,7 @@ mod tests {
             let (_, volume) = store.find_volume_mut(VolumeId(17)).unwrap();
             volume.set_read_only().unwrap();
             volume.volume_info.files.push(Default::default());
-            volume.refresh_remote_write_mode();
+            volume.refresh_remote_write_mode().unwrap();
         }
 
         let heartbeat = build_heartbeat(&test_config(), &mut store);
@@ -1944,7 +1944,7 @@ mod tests {
             key: "volumes/71.dat".to_string(),
             ..Default::default()
         });
-        volume.refresh_remote_write_mode();
+        volume.refresh_remote_write_mode().unwrap();
 
         let heartbeat = build_heartbeat(&test_config(), &mut store);
 
diff --git a/seaweed-volume/src/storage/needle_map.rs b/seaweed-volume/src/storage/needle_map.rs
index 2d0dfb456..cdbfa731a 100644
--- a/seaweed-volume/src/storage/needle_map.rs
+++ b/seaweed-volume/src/storage/needle_map.rs
@@ -14,7 +14,10 @@ use std::path::Path;
 use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};
 
 mod compact_map;
+pub mod file_pool;
+pub mod sorted_file;
 use compact_map::CompactMap;
+use sorted_file::SortedFileNeedleMap;
 
 use redb::{Database, Durability, ReadableDatabase, ReadableTable, TableDefinition};
 
@@ -939,33 +942,31 @@ impl RedbNeedleMap {
     }
 
     /// Collect all entries as a Vec for iteration (used by volume.rs iter patterns).
-    pub fn collect_entries(&self) -> Vec<(NeedleId, NeedleValue)> {
+    pub fn collect_entries(&self) -> io::Result<Vec<(NeedleId, NeedleValue)>> {
         let mut result = Vec::new();
-        let txn: redb::ReadTransaction = match self.db.begin_read() {
-            Ok(t) => t,
-            Err(_) => return result,
-        };
-        let table = match txn.open_table(NEEDLE_TABLE) {
-            Ok(t) => t,
-            Err(_) => return result,
-        };
-        let iter = match table.iter() {
-            Ok(i) => i,
-            Err(_) => return result,
-        };
+        let txn: redb::ReadTransaction = self
+            .db
+            .begin_read()
+            .map_err(|e| io::Error::other(format!("redb begin_read: {e}")))?;
+        let table = txn
+            .open_table(NEEDLE_TABLE)
+            .map_err(|e| io::Error::other(format!("redb open_table: {e}")))?;
+        let iter = table
+            .iter()
+            .map_err(|e| io::Error::other(format!("redb iter: {e}")))?;
         for entry in iter {
-            if let Ok((key_guard, val_guard)) = entry {
-                let key_u64: u64 = key_guard.value();
-                let bytes: &[u8] = val_guard.value();
-                if bytes.len() == PACKED_NEEDLE_VALUE_SIZE {
-                    let mut arr = [0u8; PACKED_NEEDLE_VALUE_SIZE];
-                    arr.copy_from_slice(bytes);
-                    let nv = unpack_needle_value(&arr);
-                    result.push((NeedleId(key_u64), nv));
-                }
+            let (key_guard, val_guard) =
+                entry.map_err(|e| io::Error::other(format!("redb entry: {e}")))?;
+            let key_u64: u64 = key_guard.value();
+            let bytes: &[u8] = val_guard.value();
+            if bytes.len() == PACKED_NEEDLE_VALUE_SIZE {
+                let mut arr = [0u8; PACKED_NEEDLE_VALUE_SIZE];
+                arr.copy_from_slice(bytes);
+                let nv = unpack_needle_value(&arr);
+                result.push((NeedleId(key_u64), nv));
             }
         }
-        result
+        Ok(result)
     }
 }
 
@@ -977,6 +978,10 @@ impl RedbNeedleMap {
 pub enum NeedleMap {
     InMemory(CompactNeedleMap),
     Redb(RedbNeedleMap),
+    /// Read-only volumes — including every cloud-tiered one — search the sorted
+    /// `.sdx` on disk instead of holding an index in RAM. Mirrors Go's
+    /// `SortedFileNeedleMap`.
+    SortedFile(SortedFileNeedleMap),
 }
 
 impl NeedleMap {
@@ -985,6 +990,7 @@ impl NeedleMap {
         match self {
             NeedleMap::InMemory(nm) => nm.put(key, offset, size),
             NeedleMap::Redb(nm) => nm.put(key, offset, size),
+            NeedleMap::SortedFile(nm) => nm.put(key, offset, size),
         }
     }
 
@@ -993,6 +999,7 @@ impl NeedleMap {
         match self {
             NeedleMap::InMemory(nm) => nm.get(key),
             NeedleMap::Redb(nm) => nm.get(key),
+            NeedleMap::SortedFile(nm) => nm.get(key),
         }
     }
 
@@ -1001,6 +1008,7 @@ impl NeedleMap {
         match self {
             NeedleMap::InMemory(nm) => nm.delete(key, offset),
             NeedleMap::Redb(nm) => nm.delete(key, offset),
+            NeedleMap::SortedFile(nm) => nm.delete(key, offset),
         }
     }
 
@@ -1009,6 +1017,9 @@ impl NeedleMap {
         match self {
             NeedleMap::InMemory(nm) => nm.set_idx_file(file, offset),
             NeedleMap::Redb(nm) => nm.set_idx_file(file, offset),
+            // The sorted map borrows its .idx per append, so there is no
+            // long-lived writer to install.
+            NeedleMap::SortedFile(_) => {}
         }
     }
 
@@ -1017,6 +1028,8 @@ impl NeedleMap {
         match self {
             NeedleMap::InMemory(nm) => nm.has_idx_writer(),
             NeedleMap::Redb(nm) => nm.has_idx_writer(),
+            // Appends open the .idx on demand, so one is always available.
+            NeedleMap::SortedFile(_) => true,
         }
     }
 
@@ -1025,6 +1038,7 @@ impl NeedleMap {
         match self {
             NeedleMap::InMemory(nm) => nm.content_size(),
             NeedleMap::Redb(nm) => nm.content_size(),
+            NeedleMap::SortedFile(nm) => nm.content_size(),
         }
     }
 
@@ -1033,6 +1047,7 @@ impl NeedleMap {
         match self {
             NeedleMap::InMemory(nm) => nm.deleted_size(),
             NeedleMap::Redb(nm) => nm.deleted_size(),
+            NeedleMap::SortedFile(nm) => nm.deleted_size(),
         }
     }
 
@@ -1041,6 +1056,7 @@ impl NeedleMap {
         match self {
             NeedleMap::InMemory(nm) => nm.file_count(),
             NeedleMap::Redb(nm) => nm.file_count(),
+            NeedleMap::SortedFile(nm) => nm.file_count(),
         }
     }
 
@@ -1049,6 +1065,7 @@ impl NeedleMap {
         match self {
             NeedleMap::InMemory(nm) => nm.deleted_count(),
             NeedleMap::Redb(nm) => nm.deleted_count(),
+            NeedleMap::SortedFile(nm) => nm.deleted_count(),
         }
     }
 
@@ -1057,6 +1074,7 @@ impl NeedleMap {
         match self {
             NeedleMap::InMemory(nm) => nm.max_file_key(),
             NeedleMap::Redb(nm) => nm.max_file_key(),
+            NeedleMap::SortedFile(nm) => nm.max_file_key(),
         }
     }
 
@@ -1067,6 +1085,7 @@ impl NeedleMap {
         match self {
             NeedleMap::InMemory(nm) => nm.max_needle_end(),
             NeedleMap::Redb(nm) => nm.max_needle_end(),
+            NeedleMap::SortedFile(nm) => nm.max_needle_end(),
         }
     }
 
@@ -1075,6 +1094,7 @@ impl NeedleMap {
         match self {
             NeedleMap::InMemory(nm) => nm.index_file_size(),
             NeedleMap::Redb(nm) => nm.index_file_size(),
+            NeedleMap::SortedFile(nm) => nm.index_file_size(),
         }
     }
 
@@ -1083,6 +1103,7 @@ impl NeedleMap {
         match self {
             NeedleMap::InMemory(nm) => nm.sync(),
             NeedleMap::Redb(nm) => nm.sync(),
+            NeedleMap::SortedFile(nm) => nm.sync(),
         }
     }
 
@@ -1091,6 +1112,7 @@ impl NeedleMap {
         match self {
             NeedleMap::InMemory(nm) => nm.close(),
             NeedleMap::Redb(nm) => nm.close(),
+            NeedleMap::SortedFile(nm) => nm.close(),
         }
     }
 
@@ -1099,6 +1121,7 @@ impl NeedleMap {
         match self {
             NeedleMap::InMemory(nm) => nm.save_to_idx(path),
             NeedleMap::Redb(nm) => nm.save_to_idx(path),
+            NeedleMap::SortedFile(nm) => nm.save_to_idx(path),
         }
     }
 
@@ -1110,22 +1133,28 @@ impl NeedleMap {
         match self {
             NeedleMap::InMemory(nm) => nm.ascending_visit(f),
             NeedleMap::Redb(nm) => nm.ascending_visit(f),
+            NeedleMap::SortedFile(nm) => nm.ascending_visit(f),
         }
     }
 
     /// Iterate all entries. Returns a Vec of (NeedleId, NeedleValue) pairs.
-    /// For InMemory this collects via ascending visit; for Redb it reads from disk.
-    pub fn iter_entries(&self) -> Vec<(NeedleId, NeedleValue)> {
+    /// For InMemory this collects via ascending visit; the disk-backed maps read
+    /// it back off disk, so a truncated .sdx or a redb read fault surfaces here
+    /// as an error. Compaction treats the result as the complete live set, so a
+    /// partial scan must never be mistaken for an empty tail.
+    pub fn iter_entries(&self) -> io::Result<Vec<(NeedleId, NeedleValue)>> {
         match self {
             NeedleMap::InMemory(nm) => {
                 let mut entries = Vec::new();
+                // The visitor never fails, so neither can this.
                 let _ = nm.ascending_visit(|id, nv| {
                     entries.push((id, *nv));
                     Ok(())
                 });
-                entries
+                Ok(entries)
             }
             NeedleMap::Redb(nm) => nm.collect_entries(),
+            NeedleMap::SortedFile(nm) => nm.iter_entries(),
         }
     }
 }
diff --git a/seaweed-volume/src/storage/needle_map/file_pool.rs b/seaweed-volume/src/storage/needle_map/file_pool.rs
new file mode 100644
index 000000000..d55a090ac
--- /dev/null
+++ b/seaweed-volume/src/storage/needle_map/file_pool.rs
@@ -0,0 +1,247 @@
+//! Bounded pool of open index-file descriptors.
+//!
+//! Read-only volumes — cloud-tiered ones above all — outnumber writable ones by
+//! orders of magnitude on a large server, and a volume that pins its `.idx` and
+//! `.sdx` for the life of the process costs two descriptors whether or not
+//! anybody reads it. At ~600K volumes per server that alone exhausts any fd
+//! limit. Neither file is needed except while a lookup is in flight, so
+//! [`SortedFileNeedleMap`](super::sorted_file::SortedFileNeedleMap) borrows them
+//! from this pool: an idle volume holds nothing, a busy one keeps its handles
+//! hot rather than paying an `open()` per needle.
+//!
+//! Mirrors Go's `weed/storage/needle_map_file_pool.go`. Handles are handed out
+//! as `Arc<File>`, so an eviction cannot close a descriptor a reader still
+//! holds — the file closes when the last borrower drops its `Arc`.
+
+use std::collections::{BTreeMap, HashMap};
+use std::fs::{File, OpenOptions};
+use std::io;
+use std::sync::{Arc, Mutex, OnceLock};
+
+/// Descriptors the pool keeps open. Matches Go's `maxPooledIndexFiles`.
+pub const MAX_POOLED_INDEX_FILES: usize = 1024;
+
+struct Entry {
+    file: Arc<File>,
+    tick: u64,
+}
+
+#[derive(Default)]
+struct Inner {
+    entries: HashMap<String, Entry>,
+    /// Recency order, oldest tick first, so eviction is a `pop_first`.
+    order: BTreeMap<u64, String>,
+    next_tick: u64,
+}
+
+pub struct IndexFilePool {
+    capacity: usize,
+    inner: Mutex<Inner>,
+}
+
+/// Writable and read-only handles for the same path are pooled separately so a
+/// read never depends on the file being openable for write — a volume served
+/// off a read-only mount still answers lookups.
+fn pool_key(path: &str, writable: bool) -> String {
+    if writable {
+        format!("{path}\0rw")
+    } else {
+        path.to_string()
+    }
+}
+
+impl IndexFilePool {
+    pub fn new(capacity: usize) -> Self {
+        IndexFilePool {
+            capacity: capacity.max(1),
+            inner: Mutex::new(Inner::default()),
+        }
+    }
+
+    /// Hand out an open handle for `path`, reusing the pooled one when there is
+    /// one. The descriptor lives as long as the returned `Arc`.
+    pub fn borrow(&self, path: &str, writable: bool) -> io::Result<Arc<File>> {
+        let key = pool_key(path, writable);
+        if let Some(file) = self.touch(&key) {
+            return Ok(file);
+        }
+
+        // Opened outside the lock: a cold open blocks on disk, and holding a
+        // process-wide mutex across it would serialize every volume's lookups.
+        let file = Arc::new(OpenOptions::new().read(true).write(writable).open(path)?);
+        Ok(self.insert(key, file))
+    }
+
+    /// Forget the pooled handles for `path`, so a later rename or delete of that
+    /// path cannot be served from a descriptor on the old inode.
+    pub fn discard(&self, path: &str) {
+        let mut inner = self.inner.lock().unwrap();
+        for key in [pool_key(path, false), pool_key(path, true)] {
+            if let Some(entry) = inner.entries.remove(&key) {
+                inner.order.remove(&entry.tick);
+            }
+        }
+    }
+
+    /// Descriptors currently pooled. Test-only visibility into the bound.
+    #[cfg(test)]
+    pub fn pooled_count(&self) -> usize {
+        self.inner.lock().unwrap().entries.len()
+    }
+
+    fn touch(&self, key: &str) -> Option<Arc<File>> {
+        let mut inner = self.inner.lock().unwrap();
+        let tick = inner.next_tick;
+        let entry = inner.entries.get_mut(key)?;
+        let file = entry.file.clone();
+        let old_tick = std::mem::replace(&mut entry.tick, tick);
+        inner.order.remove(&old_tick);
+        inner.order.insert(tick, key.to_string());
+        inner.next_tick += 1;
+        Some(file)
+    }
+
+    fn insert(&self, key: String, file: Arc<File>) -> Arc<File> {
+        let mut inner = self.inner.lock().unwrap();
+        if let Some(entry) = inner.entries.get(&key) {
+            // Another borrower opened the same path first; keep one descriptor.
+            return entry.file.clone();
+        }
+        let tick = inner.next_tick;
+        inner.next_tick += 1;
+        inner.order.insert(tick, key.clone());
+        inner.entries.insert(
+            key,
+            Entry {
+                file: file.clone(),
+                tick,
+            },
+        );
+        while inner.entries.len() > self.capacity {
+            let Some((_, oldest)) = inner.order.pop_first() else {
+                break;
+            };
+            inner.entries.remove(&oldest);
+        }
+        file
+    }
+}
+
+/// Process-wide pool shared by every read-only volume on this server.
+pub fn pooled_index_files() -> &'static IndexFilePool {
+    static POOL: OnceLock<IndexFilePool> = OnceLock::new();
+    POOL.get_or_init(|| IndexFilePool::new(MAX_POOLED_INDEX_FILES))
+}
+
+/// Descriptors this process holds on `.idx`/`.sdx` files under `dir`, read from
+/// `/proc/self/fd` where it exists and from `lsof` otherwise. `None` when
+/// neither is available, so a caller can skip rather than assert vacuously.
+#[cfg(test)]
+pub(crate) fn open_index_fds(dir: &std::path::Path) -> Option<usize> {
+    let prefix = std::fs::canonicalize(dir).unwrap_or_else(|_| dir.to_path_buf());
+    let is_index = |target: &std::path::Path| {
+        target.starts_with(&prefix)
+            && matches!(
+                target.extension().and_then(|e| e.to_str()),
+                Some("idx") | Some("sdx")
+            )
+    };
+
+    if let Ok(entries) = std::fs::read_dir("/proc/self/fd") {
+        return Some(
+            entries
+                .filter_map(|e| std::fs::read_link(e.ok()?.path()).ok())
+                .filter(|target| is_index(target))
+                .count(),
+        );
+    }
+
+    let out = std::process::Command::new("lsof")
+        .args(["-p", &std::process::id().to_string(), "-F", "n"])
+        .output()
+        .ok()?;
+    Some(
+        String::from_utf8_lossy(&out.stdout)
+            .lines()
+            .filter_map(|line| line.strip_prefix('n'))
+            .filter(|line| is_index(std::path::Path::new(line)))
+            .count(),
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Write;
+
+    fn write_file(dir: &std::path::Path, name: &str, contents: &[u8]) -> String {
+        let path = dir.join(name);
+        let mut f = File::create(&path).unwrap();
+        f.write_all(contents).unwrap();
+        path.to_str().unwrap().to_string()
+    }
+
+    #[test]
+    fn evicted_handle_stays_usable_for_its_borrower() {
+        let dir = tempfile::tempdir().unwrap();
+        let first = write_file(dir.path(), "first", b"first");
+        let second = write_file(dir.path(), "second", b"second");
+
+        let pool = IndexFilePool::new(1);
+        let borrowed = pool.borrow(&first, false).unwrap();
+
+        // Pushes the single slot over, evicting the entry still in use.
+        let _other = pool.borrow(&second, false).unwrap();
+        assert_eq!(pool.pooled_count(), 1);
+
+        let mut buf = [0u8; 5];
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::FileExt;
+            borrowed.read_exact_at(&mut buf, 0).unwrap();
+        }
+        assert_eq!(&buf, b"first");
+    }
+
+    #[test]
+    fn borrow_reuses_the_pooled_handle() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = write_file(dir.path(), "idx", b"x");
+
+        let pool = IndexFilePool::new(4);
+        let a = pool.borrow(&path, false).unwrap();
+        let b = pool.borrow(&path, false).unwrap();
+        assert!(Arc::ptr_eq(&a, &b));
+        assert_eq!(pool.pooled_count(), 1);
+
+        // A writable handle is pooled separately from the read-only one.
+        let w = pool.borrow(&path, true).unwrap();
+        assert!(!Arc::ptr_eq(&a, &w));
+        assert_eq!(pool.pooled_count(), 2);
+    }
+
+    #[test]
+    fn discard_drops_both_handles() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = write_file(dir.path(), "idx", b"x");
+
+        let pool = IndexFilePool::new(4);
+        let _r = pool.borrow(&path, false).unwrap();
+        let _w = pool.borrow(&path, true).unwrap();
+        assert_eq!(pool.pooled_count(), 2);
+
+        pool.discard(&path);
+        assert_eq!(pool.pooled_count(), 0);
+    }
+
+    #[test]
+    fn pool_stays_within_capacity() {
+        let dir = tempfile::tempdir().unwrap();
+        let pool = IndexFilePool::new(3);
+        for i in 0..10 {
+            let path = write_file(dir.path(), &format!("f{i}"), b"x");
+            let _ = pool.borrow(&path, false).unwrap();
+        }
+        assert_eq!(pool.pooled_count(), 3);
+    }
+}
diff --git a/seaweed-volume/src/storage/needle_map/sorted_file.rs b/seaweed-volume/src/storage/needle_map/sorted_file.rs
new file mode 100644
index 000000000..ba000e9c9
--- /dev/null
+++ b/seaweed-volume/src/storage/needle_map/sorted_file.rs
@@ -0,0 +1,908 @@
+//! Disk-backed needle map for read-only volumes, mirroring Go's
+//! `weed/storage/needle_map_sorted_file.go`.
+//!
+//! A read-only or cloud-tiered volume does not need its index in RAM: `.sdx` is
+//! the `.idx` rewritten as a sorted array of live entries, so a lookup is a
+//! binary search on disk. Deletes append a tombstone to the tail of `.idx` and
+//! mark the `.sdx` record in place.
+//!
+//! The map holds no descriptor of its own — both files are borrowed per
+//! operation from [`file_pool`](super::file_pool) — so a volume nobody is
+//! reading costs zero fds and zero index bytes of RAM.
+
+use std::fs::{File, OpenOptions};
+use std::io;
+use std::sync::atomic::Ordering;
+use std::sync::Mutex;
+
+use super::file_pool::pooled_index_files;
+use crate::storage::idx;
+use crate::storage::needle_map::{CompactNeedleMap, NeedleMapMetric, NeedleValue};
+use crate::storage::types::*;
+
+/// Entries per positional read when scanning `.sdx` end to end.
+const ROWS_TO_READ: u64 = 1024;
+
+/// Byte offset of the `Size` field inside a 17-byte index entry.
+const ENTRY_SIZE_OFFSET: u64 = (NEEDLE_ID_SIZE + OFFSET_SIZE) as u64;
+
+/// Appends made since the last sync, plus the tail offset the next tombstone
+/// goes to. Seeded from the `.idx` size at open so a delete appends instead of
+/// overwriting the front of the file.
+struct IndexTail {
+    offset: u64,
+    needs_sync: bool,
+}
+
+pub struct SortedFileNeedleMap {
+    metric: NeedleMapMetric,
+    index_file_name: String,
+    db_file_name: String,
+    db_file_size: i64,
+    tail: Mutex<IndexTail>,
+}
+
+impl SortedFileNeedleMap {
+    /// Open the sorted map for the volume whose index files share
+    /// `index_base_file_name` (the path with no extension), regenerating `.sdx`
+    /// when it is older than `.idx` or torn.
+    pub fn open(index_base_file_name: &str, version: Version) -> io::Result<Self> {
+        let index_file_name = format!("{index_base_file_name}.idx");
+        let db_file_name = format!("{index_base_file_name}.sdx");
+
+        if !is_sorted_file_fresh(&db_file_name, &index_file_name)
+            || !holds_whole_entries(&db_file_name)
+        {
+            tracing::info!(sdx = %db_file_name, idx = %index_file_name, "generating sorted index");
+            write_sorted_file_from_idx(&index_file_name, &db_file_name, version)?;
+        }
+
+        let db_file_size = std::fs::metadata(&db_file_name)?.len();
+        if db_file_size % NEEDLE_MAP_ENTRY_SIZE as u64 != 0 {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                format!(
+                    "{db_file_name} is {db_file_size} bytes, not whole {NEEDLE_MAP_ENTRY_SIZE}-byte entries"
+                ),
+            ));
+        }
+        let db_file_size = db_file_size as i64;
+        let index_file_size = std::fs::metadata(&index_file_name)?.len();
+        // Start appends at the last whole entry. A torn tail is overwritten by
+        // the next tombstone rather than leaving every later row misaligned; the
+        // metric walk below ignores it for the same reason.
+        let index_file_size = index_file_size - index_file_size % NEEDLE_MAP_ENTRY_SIZE as u64;
+
+        let metric = index_metric(&index_file_name, &db_file_name, version)?;
+
+        Ok(SortedFileNeedleMap {
+            metric,
+            index_file_name,
+            db_file_name,
+            db_file_size,
+            tail: Mutex::new(IndexTail {
+                offset: index_file_size,
+                needs_sync: false,
+            }),
+        })
+    }
+
+    /// Path of the `.sdx` this map searches.
+    pub fn db_file_name(&self) -> &str {
+        &self.db_file_name
+    }
+
+    /// Path of the `.idx` this map appends tombstones to.
+    pub fn index_file_name(&self) -> &str {
+        &self.index_file_name
+    }
+
+    pub fn get(&self, key: NeedleId) -> Option<NeedleValue> {
+        let file = match pooled_index_files().borrow(&self.db_file_name, false) {
+            Ok(file) => file,
+            Err(e) => {
+                tracing::warn!(sdx = %self.db_file_name, error = %e, "open sorted index");
+                return None;
+            }
+        };
+        match search_sorted_index(&file, self.db_file_size, key) {
+            Ok(Some((_, offset, size))) => Some(NeedleValue { offset, size }),
+            Ok(None) => None,
+            Err(e) => {
+                tracing::warn!(sdx = %self.db_file_name, error = %e, "search sorted index");
+                None
+            }
+        }
+    }
+
+    /// Always an error: a volume backed by `.sdx` is read-only. Mirrors Go's
+    /// `SortedFileNeedleMap.Put`.
+    pub fn put(&mut self, _key: NeedleId, _offset: Offset, _size: Size) -> io::Result<()> {
+        Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            format!("needle map {} is read only", self.db_file_name),
+        ))
+    }
+
+    /// Append a tombstone for `key` to `.idx` and mark its `.sdx` record
+    /// deleted. A key that is absent or already deleted is a no-op, matching
+    /// Go.
+    pub fn delete(&self, key: NeedleId, offset: Offset) -> io::Result<Option<Size>> {
+        let file = pooled_index_files().borrow(&self.db_file_name, true)?;
+        let Some((entry_index, _, size)) = search_sorted_index(&file, self.db_file_size, key)?
+        else {
+            return Ok(None);
+        };
+        if size.is_deleted() {
+            return Ok(None);
+        }
+
+        // Write to the index file first: .idx is the source of truth a reload
+        // rebuilds .sdx from. If the in-place mark below then fails, the delete
+        // is still durable and neither shape of failure corrupts the index: a
+        // write that lands nothing leaves .sdx older than the .idx we just
+        // appended to, so the next load rebuilds it, and a partial write leaves
+        // the size field's leading 0xff behind, which reads back negative and so
+        // already counts as deleted.
+        self.append_to_index_file(key, offset, TOMBSTONE_FILE_SIZE)?;
+
+        let mut buf = [0u8; SIZE_SIZE];
+        TOMBSTONE_FILE_SIZE.to_bytes(&mut buf);
+        write_at(
+            &file,
+            &buf,
+            entry_index * NEEDLE_MAP_ENTRY_SIZE as u64 + ENTRY_SIZE_OFFSET,
+        )?;
+
+        // Move the counters to where a reload would put them, so the heartbeat
+        // does not report this volume as garbage-free until it restarts. The
+        // tombstone is one more .idx row, and under the rule index_metric
+        // applies both it and the row it supersedes count as deletions; the
+        // bytes leave the live set without leaving the .idx.
+        self.metric.file_count.fetch_add(1, Ordering::Relaxed);
+        self.metric.deletion_count.fetch_add(2, Ordering::Relaxed);
+        self.metric
+            .deletion_byte_count
+            .fetch_add(size.0.max(0) as u64, Ordering::Relaxed);
+        Ok(Some(size))
+    }
+
+    fn append_to_index_file(&self, key: NeedleId, offset: Offset, size: Size) -> io::Result<()> {
+        let file = pooled_index_files().borrow(&self.index_file_name, true)?;
+        let mut buf = [0u8; NEEDLE_MAP_ENTRY_SIZE];
+        idx_entry_to_bytes(&mut buf, key, offset, size);
+
+        let mut tail = self.tail.lock().unwrap();
+        write_at(&file, &buf, tail.offset)?;
+        tail.offset += NEEDLE_MAP_ENTRY_SIZE as u64;
+        tail.needs_sync = true;
+        Ok(())
+    }
+
+    /// Answers from the offset the appends maintain rather than a stat: the
+    /// heartbeat asks every volume for this on every beat, and a read-only
+    /// volume's `.idx` only ever grows through `append_to_index_file`.
+    pub fn index_file_size(&self) -> u64 {
+        self.tail.lock().unwrap().offset
+    }
+
+    /// Flushes tombstones appended by [`delete`](Self::delete). A read-only
+    /// volume that has never been deleted from — the overwhelming majority —
+    /// opens nothing here, so shutting down a server holding hundreds of
+    /// thousands of them costs no fsyncs.
+    pub fn sync(&self) -> io::Result<()> {
+        let mut tail = self.tail.lock().unwrap();
+        if !tail.needs_sync {
+            return Ok(());
+        }
+        let file = pooled_index_files().borrow(&self.index_file_name, true)?;
+        file.sync_all()?;
+        tail.needs_sync = false;
+        Ok(())
+    }
+
+    /// Drops the pooled handles: the caller may be about to rename or remove
+    /// these paths, and a descriptor left behind would keep answering reads
+    /// from the old inode.
+    pub fn close(&mut self) {
+        let _ = self.sync();
+        pooled_index_files().discard(&self.index_file_name);
+        pooled_index_files().discard(&self.db_file_name);
+    }
+
+    /// Visit the live entries in ascending needle-id order, straight off `.sdx`.
+    fn visit_live_entries<F>(&self, mut f: F) -> io::Result<()>
+    where
+        F: FnMut(NeedleId, &NeedleValue) -> io::Result<()>,
+    {
+        let file = pooled_index_files().borrow(&self.db_file_name, false)?;
+        let entry_count = self.db_file_size.max(0) as u64 / NEEDLE_MAP_ENTRY_SIZE as u64;
+        // A batch per 1024 entries rather than a syscall per entry, matching
+        // idx::walk_index_file. Positional reads, not a cursor: the handle is
+        // shared with any other borrower, so a file position would race.
+        let rows_per_read = ROWS_TO_READ.min(entry_count.max(1));
+        let mut block = vec![0u8; rows_per_read as usize * NEEDLE_MAP_ENTRY_SIZE];
+        let mut done: u64 = 0;
+        while done < entry_count {
+            let rows = rows_per_read.min(entry_count - done) as usize;
+            let bytes = &mut block[..rows * NEEDLE_MAP_ENTRY_SIZE];
+            read_exact_at(&file, bytes, done * NEEDLE_MAP_ENTRY_SIZE as u64)?;
+            for entry in bytes.chunks_exact(NEEDLE_MAP_ENTRY_SIZE) {
+                let (key, offset, size) = idx_entry_from_bytes(entry);
+                if !size.is_valid() {
+                    continue; // deleted in place by a runtime delete
+                }
+                f(key, &NeedleValue { offset, size })?;
+            }
+            done += rows as u64;
+        }
+        Ok(())
+    }
+
+    pub fn ascending_visit<F>(&self, mut f: F) -> Result<(), String>
+    where
+        F: FnMut(NeedleId, &NeedleValue) -> Result<(), String>,
+    {
+        let mut visit_error = None;
+        self.visit_live_entries(|id, nv| {
+            if let Err(e) = f(id, nv) {
+                visit_error = Some(e);
+                return Err(io::Error::other("visit aborted"));
+            }
+            Ok(())
+        })
+        .map_err(|e| visit_error.take().unwrap_or_else(|| e.to_string()))
+    }
+
+    pub fn iter_entries(&self) -> io::Result<Vec<(NeedleId, NeedleValue)>> {
+        let mut entries = Vec::new();
+        self.visit_live_entries(|id, nv| {
+            entries.push((id, *nv));
+            Ok(())
+        })?;
+        Ok(entries)
+    }
+
+    pub fn save_to_idx(&self, path: &str) -> io::Result<()> {
+        let mut out = OpenOptions::new()
+            .write(true)
+            .create(true)
+            .truncate(true)
+            .open(path)?;
+        self.visit_live_entries(|id, nv| idx::write_index_entry(&mut out, id, nv.offset, nv.size))?;
+        out.sync_all()
+    }
+
+    // ---- Metrics accessors ----
+
+    pub fn content_size(&self) -> u64 {
+        self.metric.file_byte_count.load(Ordering::Relaxed)
+    }
+
+    pub fn deleted_size(&self) -> u64 {
+        self.metric.deletion_byte_count.load(Ordering::Relaxed)
+    }
+
+    pub fn file_count(&self) -> i64 {
+        self.metric.file_count.load(Ordering::Relaxed)
+    }
+
+    pub fn deleted_count(&self) -> i64 {
+        self.metric.deletion_count.load(Ordering::Relaxed)
+    }
+
+    pub fn max_file_key(&self) -> NeedleId {
+        NeedleId(self.metric.max_file_key.load(Ordering::Relaxed))
+    }
+
+    pub fn max_needle_end(&self) -> i64 {
+        self.metric.max_needle_end.load(Ordering::Relaxed)
+    }
+}
+
+impl Drop for SortedFileNeedleMap {
+    fn drop(&mut self) {
+        self.close();
+    }
+}
+
+/// Whether a sorted index is a whole number of entries. A torn one is treated
+/// like a stale one and rebuilt: truncation bumps the mtime, so the staleness
+/// check alone would trust it, and flooring the entry count would silently hide
+/// the last needle — which a later compaction would then drop for good. Go
+/// writes `.sdx` in place, so a crash mid-generation can leave one behind.
+fn holds_whole_entries(db_file_name: &str) -> bool {
+    std::fs::metadata(db_file_name)
+        .map(|m| m.len() % NEEDLE_MAP_ENTRY_SIZE as u64 == 0)
+        .unwrap_or(false)
+}
+
+/// `.sdx` is stale when it is not newer than `.idx`; writes always land in
+/// `.idx` first. Mirrors Go's `isSortedFileFresh`.
+fn is_sorted_file_fresh(db_file_name: &str, index_file_name: &str) -> bool {
+    let (Ok(db), Ok(index)) = (
+        std::fs::metadata(db_file_name),
+        std::fs::metadata(index_file_name),
+    ) else {
+        return false;
+    };
+    match (db.modified(), index.modified()) {
+        (Ok(db_time), Ok(index_time)) => db_time > index_time,
+        _ => false,
+    }
+}
+
+/// Rewrite `.idx` as a sorted array of the entries that are still live, the
+/// same file Go's `WriteSortedFileFromIdx` produces: last write wins per key,
+/// and a tombstoned key is dropped entirely.
+fn write_sorted_file_from_idx(
+    index_file_name: &str,
+    db_file_name: &str,
+    version: Version,
+) -> io::Result<()> {
+    let mut index_file = File::open(index_file_name)?;
+    let nm = CompactNeedleMap::load_from_idx(&mut index_file, version)?;
+
+    let tmp_name = format!("{db_file_name}.tmp");
+    let mut out = OpenOptions::new()
+        .write(true)
+        .create(true)
+        .truncate(true)
+        .open(&tmp_name)?;
+    let write_result = nm
+        .ascending_visit(|id, nv| {
+            idx::write_index_entry(&mut out, id, nv.offset, nv.size).map_err(|e| e.to_string())
+        })
+        .map_err(io::Error::other)
+        .and_then(|()| out.sync_all());
+    if let Err(e) = write_result {
+        let _ = std::fs::remove_file(&tmp_name);
+        return Err(e);
+    }
+    drop(out);
+    // Rename in: a crash mid-generation must not leave a short .sdx that looks
+    // fresh and silently hides needles.
+    std::fs::rename(&tmp_name, db_file_name)
+}
+
+/// Counters equivalent to Go's `needleMapMetricFromIndexFile`, derived without
+/// its bloom filter — and so without its false positives or its per-volume
+/// allocation.
+///
+/// Go counts every `.idx` row in `FileCounter`, and counts a row in
+/// `DeletionCounter` when it is not the last, still-valid row for its key. The
+/// rows that *are* the last valid row for their key are exactly the entries
+/// `.sdx` holds, so `DeletionCounter = rows - live` and
+/// `DeletionByteCounter = valid bytes in .idx - live bytes in .sdx`, each
+/// computable in one sequential pass with no per-key state.
+fn index_metric(
+    index_file_name: &str,
+    db_file_name: &str,
+    version: Version,
+) -> io::Result<NeedleMapMetric> {
+    let metric = NeedleMapMetric::default();
+
+    let mut rows: i64 = 0;
+    let mut valid_bytes: u64 = 0;
+    let mut index_file = File::open(index_file_name)?;
+    idx::walk_index_file(&mut index_file, 0, |key, offset, size| {
+        rows += 1;
+        metric.maybe_set_max_file_key(key);
+        metric.maybe_set_max_needle_end(offset, size, version);
+        if size.is_valid() {
+            valid_bytes += size.0 as u64;
+        }
+        Ok(())
+    })?;
+
+    let mut live: i64 = 0;
+    let mut live_bytes: u64 = 0;
+    let mut db_file = File::open(db_file_name)?;
+    idx::walk_index_file(&mut db_file, 0, |_, _, size| {
+        if size.is_valid() {
+            live += 1;
+            live_bytes += size.0 as u64;
+        }
+        Ok(())
+    })?;
+
+    metric.file_count.store(rows, Ordering::Relaxed);
+    metric.file_byte_count.store(valid_bytes, Ordering::Relaxed);
+    metric
+        .deletion_count
+        .store((rows - live).max(0), Ordering::Relaxed);
+    metric
+        .deletion_byte_count
+        .store(valid_bytes.saturating_sub(live_bytes), Ordering::Relaxed);
+    Ok(metric)
+}
+
+/// Binary search the sorted index for `key`, returning its entry index along
+/// with the record. Mirrors Go's `SearchNeedleFromSortedIndex`.
+fn search_sorted_index(
+    file: &File,
+    file_size: i64,
+    key: NeedleId,
+) -> io::Result<Option<(u64, Offset, Size)>> {
+    let mut lo: u64 = 0;
+    let mut hi: u64 = file_size.max(0) as u64 / NEEDLE_MAP_ENTRY_SIZE as u64;
+    let mut buf = [0u8; NEEDLE_MAP_ENTRY_SIZE];
+    while lo < hi {
+        let mid = lo + (hi - lo) / 2;
+        read_exact_at(file, &mut buf, mid * NEEDLE_MAP_ENTRY_SIZE as u64)?;
+        let (entry_key, offset, size) = idx_entry_from_bytes(&buf);
+        if entry_key == key {
+            return Ok(Some((mid, offset, size)));
+        }
+        if entry_key < key {
+            lo = mid + 1;
+        } else {
+            hi = mid;
+        }
+    }
+    Ok(None)
+}
+
+fn read_exact_at(file: &File, buf: &mut [u8], offset: u64) -> io::Result<()> {
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::FileExt;
+        file.read_exact_at(buf, offset)
+    }
+    #[cfg(windows)]
+    {
+        use std::os::windows::fs::FileExt;
+        let mut filled = 0;
+        let mut at = offset;
+        while filled < buf.len() {
+            let n = file.seek_read(&mut buf[filled..], at)?;
+            if n == 0 {
+                return Err(io::Error::new(
+                    io::ErrorKind::UnexpectedEof,
+                    "unexpected EOF in seek_read",
+                ));
+            }
+            filled += n;
+            at += n as u64;
+        }
+        Ok(())
+    }
+}
+
+fn write_at(file: &File, buf: &[u8], offset: u64) -> io::Result<()> {
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::FileExt;
+        file.write_all_at(buf, offset)
+    }
+    #[cfg(windows)]
+    {
+        use std::os::windows::fs::FileExt;
+        let mut written = 0;
+        let mut at = offset;
+        while written < buf.len() {
+            let n = file.seek_write(&buf[written..], at)?;
+            if n == 0 {
+                return Err(io::Error::new(
+                    io::ErrorKind::WriteZero,
+                    "seek_write wrote nothing",
+                ));
+            }
+            written += n;
+            at += n as u64;
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Write as _;
+
+    fn version() -> Version {
+        VERSION_3
+    }
+
+    /// Write an .idx with the given rows, in order.
+    fn write_idx(path: &str, rows: &[(u64, u32, i32)]) {
+        let mut f = File::create(path).unwrap();
+        for &(key, offset, size) in rows {
+            let mut buf = [0u8; NEEDLE_MAP_ENTRY_SIZE];
+            idx_entry_to_bytes(
+                &mut buf,
+                NeedleId(key),
+                Offset::from_actual_offset(offset as i64),
+                Size(size),
+            );
+            f.write_all(&buf).unwrap();
+        }
+        f.sync_all().unwrap();
+    }
+
+    fn base(dir: &tempfile::TempDir) -> String {
+        dir.path().join("7").to_str().unwrap().to_string()
+    }
+
+    #[test]
+    fn get_finds_live_needles_and_skips_deleted_ones() {
+        let dir = tempfile::tempdir().unwrap();
+        let base = base(&dir);
+        write_idx(
+            &format!("{base}.idx"),
+            &[
+                (1, 8, 100),
+                (2, 16, 200),
+                (3, 24, 300),
+                (2, 0, -1), // tombstone
+            ],
+        );
+
+        let m = SortedFileNeedleMap::open(&base, version()).unwrap();
+        assert_eq!(m.get(NeedleId(1)).unwrap().size, Size(100));
+        assert_eq!(m.get(NeedleId(3)).unwrap().size, Size(300));
+        // A key tombstoned before the .sdx was written is not in it at all.
+        assert!(m.get(NeedleId(2)).is_none());
+        assert!(m.get(NeedleId(99)).is_none());
+    }
+
+    #[test]
+    fn metrics_match_the_go_counters() {
+        let dir = tempfile::tempdir().unwrap();
+        let base = base(&dir);
+        // key 1 rewritten once, key 2 deleted, key 3 live.
+        write_idx(
+            &format!("{base}.idx"),
+            &[
+                (1, 8, 100),
+                (2, 16, 200),
+                (1, 32, 150),
+                (3, 24, 300),
+                (2, 0, -1),
+            ],
+        );
+
+        let m = SortedFileNeedleMap::open(&base, version()).unwrap();
+        // Every row counts, matching Go's FileCounter.
+        assert_eq!(m.file_count(), 5);
+        assert_eq!(m.content_size(), 100 + 200 + 150 + 300);
+        // Superseded row for key 1, plus both rows for the deleted key 2.
+        assert_eq!(m.deleted_count(), 3);
+        assert_eq!(m.deleted_size(), 100 + 200);
+        assert_eq!(m.max_file_key(), NeedleId(3));
+    }
+
+    #[test]
+    fn put_is_rejected() {
+        let dir = tempfile::tempdir().unwrap();
+        let base = base(&dir);
+        write_idx(&format!("{base}.idx"), &[(1, 8, 100)]);
+
+        let mut m = SortedFileNeedleMap::open(&base, version()).unwrap();
+        let err = m
+            .put(NeedleId(2), Offset::from_actual_offset(16), Size(10))
+            .unwrap_err();
+        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
+    }
+
+    #[test]
+    fn delete_appends_a_tombstone_to_the_idx_tail() {
+        let dir = tempfile::tempdir().unwrap();
+        let base = base(&dir);
+        let idx_path = format!("{base}.idx");
+        write_idx(&idx_path, &[(1, 8, 100), (2, 16, 200), (3, 24, 300)]);
+        let before = std::fs::metadata(&idx_path).unwrap().len();
+
+        let m = SortedFileNeedleMap::open(&base, version()).unwrap();
+        assert_eq!(
+            m.delete(NeedleId(2), Offset::from_actual_offset(16))
+                .unwrap(),
+            Some(Size(200))
+        );
+
+        // Appended, not overwritten from the front.
+        let after = std::fs::metadata(&idx_path).unwrap().len();
+        assert_eq!(after, before + NEEDLE_MAP_ENTRY_SIZE as u64);
+        assert_eq!(m.index_file_size(), after);
+
+        let mut rows = Vec::new();
+        let mut f = File::open(&idx_path).unwrap();
+        idx::walk_index_file(&mut f, 0, |key, _, size| {
+            rows.push((key, size));
+            Ok(())
+        })
+        .unwrap();
+        assert_eq!(rows[0], (NeedleId(1), Size(100)));
+        assert_eq!(rows[1], (NeedleId(2), Size(200)));
+        assert_eq!(rows[2], (NeedleId(3), Size(300)));
+        assert_eq!(rows[3], (NeedleId(2), TOMBSTONE_FILE_SIZE));
+
+        // .sdx is marked in place, so the needle reads back as a tombstone
+        // without a reload — the same contract Go's Get has, where callers
+        // check size.is_deleted().
+        assert!(m.get(NeedleId(2)).unwrap().size.is_deleted());
+        assert!(m
+            .delete(NeedleId(2), Offset::from_actual_offset(16))
+            .unwrap()
+            .is_none());
+        assert!(!m.get(NeedleId(1)).unwrap().size.is_deleted());
+    }
+
+    #[test]
+    fn delete_moves_the_counters_where_a_reload_would() {
+        let dir = tempfile::tempdir().unwrap();
+        let base = base(&dir);
+        write_idx(
+            &format!("{base}.idx"),
+            &[(1, 8, 100), (2, 16, 200), (3, 24, 300)],
+        );
+
+        let m = SortedFileNeedleMap::open(&base, version()).unwrap();
+        assert_eq!(
+            (m.file_count(), m.deleted_count(), m.deleted_size()),
+            (3, 0, 0)
+        );
+
+        m.delete(NeedleId(2), Offset::from_actual_offset(16))
+            .unwrap();
+        let live = m.file_count() - m.deleted_count();
+        assert_eq!(live, 2, "one of the three needles is gone");
+        assert_eq!(m.deleted_size(), 200);
+        assert_eq!(m.content_size(), 600, "a delete does not shrink .idx bytes");
+
+        // A reload recomputes from .idx and .sdx; the runtime numbers must
+        // already agree with it or the heartbeat jumps at every restart.
+        drop(m);
+        let reloaded = SortedFileNeedleMap::open(&base, version()).unwrap();
+        assert_eq!(reloaded.file_count(), 4);
+        assert_eq!(reloaded.deleted_count(), 2);
+        assert_eq!(reloaded.deleted_size(), 200);
+        assert_eq!(reloaded.content_size(), 600);
+
+        // Deleting again is a no-op and must not move them further.
+        reloaded
+            .delete(NeedleId(2), Offset::from_actual_offset(16))
+            .unwrap();
+        assert_eq!((reloaded.file_count(), reloaded.deleted_count()), (4, 2));
+    }
+
+    #[test]
+    fn a_torn_idx_tail_does_not_misalign_appends() {
+        let dir = tempfile::tempdir().unwrap();
+        let base = base(&dir);
+        let idx_path = format!("{base}.idx");
+        write_idx(&idx_path, &[(1, 8, 100), (2, 16, 200)]);
+        // A short write left half a record behind. Seeding the append offset
+        // from the raw file length would put every later row off by five bytes.
+        {
+            use std::io::Write as _;
+            let mut f = OpenOptions::new().append(true).open(&idx_path).unwrap();
+            f.write_all(&[0xab; 5]).unwrap();
+        }
+
+        let m = SortedFileNeedleMap::open(&base, version()).unwrap();
+        assert_eq!(m.index_file_size(), 2 * NEEDLE_MAP_ENTRY_SIZE as u64);
+        m.delete(NeedleId(1), Offset::from_actual_offset(8))
+            .unwrap();
+
+        let mut rows = Vec::new();
+        let mut f = File::open(&idx_path).unwrap();
+        idx::walk_index_file(&mut f, 0, |key, _, size| {
+            rows.push((key, size));
+            Ok(())
+        })
+        .unwrap();
+        assert_eq!(
+            std::fs::metadata(&idx_path).unwrap().len(),
+            3 * NEEDLE_MAP_ENTRY_SIZE as u64,
+            "the tombstone should have overwritten the torn tail"
+        );
+        assert_eq!(rows[0], (NeedleId(1), Size(100)));
+        assert_eq!(rows[1], (NeedleId(2), Size(200)));
+        assert_eq!(rows[2], (NeedleId(1), TOMBSTONE_FILE_SIZE));
+    }
+
+    #[test]
+    fn torn_sdx_is_regenerated() {
+        let dir = tempfile::tempdir().unwrap();
+        let base = base(&dir);
+        write_idx(
+            &format!("{base}.idx"),
+            &[(1, 8, 100), (2, 16, 200), (3, 24, 300)],
+        );
+
+        // Build a good .sdx, then lop off part of its last record. Truncation
+        // bumps the mtime, so the staleness check sees a *fresher* file than the
+        // .idx — without the whole-entry check the map would trust it and needle
+        // 3 would disappear.
+        drop(SortedFileNeedleMap::open(&base, version()).unwrap());
+        let good = std::fs::metadata(format!("{base}.sdx")).unwrap().len();
+        assert_eq!(good, 3 * NEEDLE_MAP_ENTRY_SIZE as u64);
+        let sdx = OpenOptions::new()
+            .write(true)
+            .open(format!("{base}.sdx"))
+            .unwrap();
+        sdx.set_len(good - 5).unwrap();
+        drop(sdx);
+        pooled_index_files().discard(&format!("{base}.sdx"));
+        assert!(
+            std::fs::metadata(format!("{base}.sdx"))
+                .unwrap()
+                .modified()
+                .unwrap()
+                > std::fs::metadata(format!("{base}.idx"))
+                    .unwrap()
+                    .modified()
+                    .unwrap(),
+            "the torn file must look fresh, or the test proves nothing"
+        );
+
+        let m = SortedFileNeedleMap::open(&base, version()).unwrap();
+        assert_eq!(
+            std::fs::metadata(format!("{base}.sdx")).unwrap().len(),
+            good
+        );
+        assert_eq!(m.get(NeedleId(3)).unwrap().size, Size(300));
+        assert_eq!(m.iter_entries().unwrap().len(), 3);
+    }
+
+    #[test]
+    fn stale_sdx_is_regenerated() {
+        let dir = tempfile::tempdir().unwrap();
+        let base = base(&dir);
+        write_idx(&format!("{base}.idx"), &[(1, 8, 100)]);
+        // A short, stale .sdx from an older .idx must not be trusted.
+        std::fs::write(format!("{base}.sdx"), b"").unwrap();
+        filetime_backdate(&format!("{base}.sdx"));
+
+        let m = SortedFileNeedleMap::open(&base, version()).unwrap();
+        assert_eq!(m.get(NeedleId(1)).unwrap().size, Size(100));
+    }
+
+    #[test]
+    fn scan_spans_read_batches() {
+        let dir = tempfile::tempdir().unwrap();
+        let base = base(&dir);
+        // Straddles ROWS_TO_READ so the batched scan has to stitch a full
+        // batch, a second full batch and a short tail together.
+        let total = ROWS_TO_READ as u32 * 2 + 7;
+        let rows: Vec<(u64, u32, i32)> = (1..=total)
+            .map(|i| (i as u64, i * 8, (i * 10) as i32))
+            .collect();
+        write_idx(&format!("{base}.idx"), &rows);
+
+        let m = SortedFileNeedleMap::open(&base, version()).unwrap();
+        let entries = m.iter_entries().unwrap();
+        assert_eq!(entries.len(), total as usize);
+        assert_eq!(entries[0].0, NeedleId(1));
+        assert_eq!(entries[entries.len() - 1].0, NeedleId(total as u64));
+        assert_eq!(entries[entries.len() - 1].1.size, Size((total * 10) as i32));
+        // Every id is still reachable through the binary search.
+        assert_eq!(
+            m.get(NeedleId(total as u64)).unwrap().size,
+            Size((total * 10) as i32)
+        );
+    }
+
+    #[test]
+    fn iter_entries_reports_a_truncated_sdx() {
+        let dir = tempfile::tempdir().unwrap();
+        let base = base(&dir);
+        write_idx(
+            &format!("{base}.idx"),
+            &[(1, 8, 100), (2, 16, 200), (3, 24, 300), (4, 32, 400)],
+        );
+
+        let m = SortedFileNeedleMap::open(&base, version()).unwrap();
+        assert_eq!(m.iter_entries().unwrap().len(), 4);
+
+        // Losing the tail of .sdx must surface as an error, not as a shorter
+        // list: compaction takes this vector for the complete live set and would
+        // otherwise commit a volume missing the needles it could not read.
+        let sdx = OpenOptions::new()
+            .write(true)
+            .open(format!("{base}.sdx"))
+            .unwrap();
+        sdx.set_len(2 * NEEDLE_MAP_ENTRY_SIZE as u64).unwrap();
+        drop(sdx);
+        pooled_index_files().discard(m.db_file_name());
+
+        let err = m.iter_entries().unwrap_err();
+        assert_eq!(err.kind(), io::ErrorKind::UnexpectedEof);
+        assert!(m.save_to_idx(&format!("{base}.check")).is_err());
+    }
+
+    #[test]
+    fn holds_no_descriptors_when_idle() {
+        let dir = tempfile::tempdir().unwrap();
+        let base = base(&dir);
+        write_idx(&format!("{base}.idx"), &[(1, 8, 100), (2, 16, 200)]);
+
+        let m = SortedFileNeedleMap::open(&base, version()).unwrap();
+        let Some(_) = super::super::file_pool::open_index_fds(dir.path()) else {
+            return; // cannot enumerate descriptors here
+        };
+        drop_pooled(&m);
+        assert_eq!(
+            super::super::file_pool::open_index_fds(dir.path()),
+            Some(0),
+            "an idle sorted map must hold no .idx/.sdx descriptor"
+        );
+
+        assert!(m.get(NeedleId(1)).is_some());
+        drop_pooled(&m);
+        assert_eq!(
+            super::super::file_pool::open_index_fds(dir.path()),
+            Some(0),
+            "a lookup must give its borrowed handle back"
+        );
+    }
+
+    fn drop_pooled(m: &SortedFileNeedleMap) {
+        pooled_index_files().discard(m.index_file_name());
+        pooled_index_files().discard(m.db_file_name());
+    }
+
+    fn filetime_backdate(path: &str) {
+        // Push the mtime into the past so the freshness check sees it as stale
+        // even when both files were written within the same clock tick.
+        let past = std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1);
+        let f = OpenOptions::new().write(true).open(path).unwrap();
+        let _ = f.set_times(std::fs::FileTimes::new().set_modified(past));
+    }
+}
+
+// Only meaningful with 5-byte offsets, which is what production Go builds use.
+#[cfg(all(test, feature = "5bytes"))]
+mod go_parity_tests {
+    use super::*;
+
+    /// The `.sdx` bytes Go's `WriteSortedFileFromIdx` produces for the fixture
+    /// below, captured from `weed/storage/erasure_coding` built with
+    /// `-tags 5BytesOffset`. A volume moved between a Go and a Rust server reads
+    /// whichever `.sdx` is already on disk, so the two generators must agree
+    /// byte for byte: sorted by needle id, last write wins, tombstoned keys
+    /// dropped entirely.
+    const GO_SDX_HEX: &str = "0000000000000001000000200000000096000000000000000300000018000000012c00000000000000040000003000000001900000000000000005000000080000\
+0001f4";
+
+    #[test]
+    fn sdx_matches_the_bytes_go_writes() {
+        let dir = tempfile::tempdir().unwrap();
+        let base = dir.path().join("1").to_str().unwrap().to_string();
+
+        // Rewrites, tombstones and out-of-order keys, matching the Go fixture.
+        let rows: &[(u64, u32, i32)] = &[
+            (5, 8, 500),
+            (1, 16, 100),
+            (3, 24, 300),
+            (1, 32, 150),
+            (2, 40, 200),
+            (2, 0, -1),
+            (4, 48, 400),
+            (9, 56, 900),
+            (9, 0, -1),
+        ];
+        let mut f = std::fs::File::create(format!("{base}.idx")).unwrap();
+        for &(key, offset, size) in rows {
+            let mut buf = [0u8; NEEDLE_MAP_ENTRY_SIZE];
+            // The Go fixture stores raw offsets; scale by the padding so both
+            // sides put the same bytes in the entry.
+            idx_entry_to_bytes(
+                &mut buf,
+                NeedleId(key),
+                Offset::from_actual_offset(offset as i64 * NEEDLE_PADDING_SIZE as i64),
+                Size(size),
+            );
+            use std::io::Write as _;
+            f.write_all(&buf).unwrap();
+        }
+        drop(f);
+
+        write_sorted_file_from_idx(&format!("{base}.idx"), &format!("{base}.sdx"), VERSION_3)
+            .unwrap();
+        let sdx = std::fs::read(format!("{base}.sdx")).unwrap();
+        let hex: String = sdx.iter().map(|b| format!("{b:02x}")).collect();
+        assert_eq!(hex, GO_SDX_HEX.replace('\n', ""));
+    }
+}
diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs
index 22c1a58ac..0c719764b 100644
--- a/seaweed-volume/src/storage/volume.rs
+++ b/seaweed-volume/src/storage/volume.rs
@@ -22,6 +22,7 @@ use tracing::{error, info, warn};
 #[cfg(test)]
 use crate::storage::idx;
 use crate::storage::needle::needle::{self, get_actual_size, Needle, NeedleError};
+use crate::storage::needle_map::sorted_file::SortedFileNeedleMap;
 use crate::storage::needle_map::{CompactNeedleMap, NeedleMap, NeedleMapKind, RedbNeedleMap};
 use crate::storage::super_block::{ReplicaPlacement, SuperBlock, SUPER_BLOCK_SIZE};
 use crate::storage::types::*;
@@ -885,7 +886,12 @@ impl Volume {
             fs::create_dir_all(parent)?;
         }
 
-        if use_redb {
+        // Go picks the sorted-file map for both read-only modes, which is every
+        // cloud-tiered volume: the index stays on disk in .sdx instead of in
+        // RAM, and nothing is held open between lookups.
+        if self.use_sorted_index() {
+            self.load_index_sorted_file(&idx_path)?;
+        } else if use_redb {
             self.load_index_redb(&idx_path)?;
         } else {
             self.load_index_inmemory(&idx_path)?;
@@ -894,6 +900,72 @@ impl Volume {
         Ok(())
     }
 
+    /// Whether this volume's index belongs in a `.sdx`. Mirrors Go's
+    /// `v.noWriteOrDelete || v.noWriteCanDelete` branch in volume_loading.go —
+    /// a tiered volume is always the second of the two.
+    fn use_sorted_index(&self) -> bool {
+        self.no_write_or_delete || self.no_write_can_delete
+    }
+
+    /// Load the on-disk sorted index for a read-only or cloud-tiered volume.
+    fn load_index_sorted_file(&mut self, idx_path: &str) -> Result<(), VolumeError> {
+        if !Path::new(idx_path).exists() {
+            // Missing .idx with existing .dat could orphan needles
+            let dat_path = self.file_name(".dat");
+            if Path::new(&dat_path).exists() {
+                let dat_size = fs::metadata(&dat_path).map(|m| m.len()).unwrap_or(0);
+                if dat_size > SUPER_BLOCK_SIZE as u64 {
+                    warn!(
+                        volume_id = self.id.0,
+                        ".idx file missing but .dat exists with data; needles may be orphaned"
+                    );
+                }
+            }
+            // Go opens .idx with O_CREATE only where deletes are allowed; an
+            // empty index yields an empty .sdx and a volume that still mounts.
+            // A volume that takes neither writes nor deletes must not write
+            // here, since its index directory can sit on a read-only mount.
+            if !self.no_write_or_delete {
+                File::create(idx_path)?;
+            }
+        }
+        // Building .sdx writes to the index directory. Where that cannot be
+        // done — a read-only mount, a full disk — keep this volume's index in
+        // memory rather than failing the load and taking its data offline.
+        match SortedFileNeedleMap::open(&self.index_file_name(), self.version()) {
+            Ok(nm) => {
+                self.nm = Some(NeedleMap::SortedFile(nm));
+                Ok(())
+            }
+            Err(e) => {
+                warn!(
+                    volume_id = self.id.0,
+                    error = %e,
+                    "cannot build the sorted index; keeping this volume's index in memory"
+                );
+                match self.load_index_inmemory(idx_path) {
+                    Ok(()) => Ok(()),
+                    // Where deletes are allowed the in-memory loader opens .idx
+                    // read-write, which fails on the same directory that just
+                    // refused the .sdx. Give up the deletes rather than the
+                    // volume: without a writer no tombstone could be recorded
+                    // anyway, so refusing them is the honest answer, and a
+                    // remount on a writable directory restores them.
+                    Err(retry_err) if !self.no_write_or_delete => {
+                        warn!(
+                            volume_id = self.id.0,
+                            error = %retry_err,
+                            "cannot open the index for write; serving this volume without deletes"
+                        );
+                        self.no_write_or_delete = true;
+                        self.load_index_inmemory(idx_path)
+                    }
+                    Err(retry_err) => Err(retry_err),
+                }
+            }
+        }
+    }
+
     /// Load index using in-memory CompactNeedleMap.
     fn load_index_inmemory(&mut self, idx_path: &str) -> Result<(), VolumeError> {
         if self.no_write_or_delete {
@@ -924,7 +996,7 @@ impl Volume {
                 .create(true)
                 .open(&idx_path)?;
 
-            let idx_size = idx_file.metadata()?.len();
+            let idx_size = trim_torn_idx_tail(&idx_file, idx_path)?;
             let mut idx_reader = io::BufReader::new(&idx_file);
             let mut nm = CompactNeedleMap::load_from_idx(&mut idx_reader, self.version())?;
 
@@ -973,7 +1045,7 @@ impl Volume {
                 .create(true)
                 .open(&idx_path)?;
 
-            let idx_size = idx_file.metadata()?.len();
+            let idx_size = trim_torn_idx_tail(&idx_file, idx_path)?;
             let mut idx_reader = io::BufReader::new(&idx_file);
             let mut nm = RedbNeedleMap::load_from_idx(&rdb_path, &mut idx_reader, self.version())?;
 
@@ -2489,7 +2561,7 @@ impl Volume {
                 .append(true)
                 .create(true)
                 .open(&idx_path)?;
-            let idx_size = write_file.metadata()?.len();
+            let idx_size = trim_torn_idx_tail(&write_file, &idx_path)?;
             if let Some(ref mut nm) = self.nm {
                 nm.set_idx_file(Box::new(write_file), idx_size);
             }
@@ -2505,17 +2577,59 @@ impl Volume {
     /// surviving until the next restart, then vanishing. Re-attach a writer
     /// here so writes persist again.
     pub fn set_writable(&mut self) -> Result<(), VolumeError> {
-        self.attach_idx_writer_if_missing()?;
+        let was_no_write_or_delete = self.no_write_or_delete;
+        let was_no_write_can_delete = self.no_write_can_delete;
         self.no_write_or_delete = false;
         // Remote-tiered volumes must stay no_write_can_delete regardless of marks.
         if !self.has_remote_file {
             self.no_write_can_delete = false;
         }
+
+        // Everything below keys off the flags above, so it runs with the volume
+        // already marked writable. Any failure has to put them back: a volume
+        // advertising writable whose needle map cannot reach .idx acknowledges
+        // writes that are gone after a restart.
+        if let Err(e) = self.publish_writable() {
+            self.no_write_or_delete = was_no_write_or_delete;
+            self.no_write_can_delete = was_no_write_can_delete;
+            return Err(e);
+        }
+        Ok(())
+    }
+
+    /// The steps that must all succeed before a volume can be left marked
+    /// writable: swap out the read-only map, attach the .idx writer, persist the
+    /// mark. Split out so set_writable has one rollback point rather than three.
+    fn publish_writable(&mut self) -> Result<(), VolumeError> {
+        self.reopen_idx_for_write()?;
+        self.attach_idx_writer_if_missing()?;
         self.save_vif()
     }
 
-    /// Recompute the Go-style write/delete mode from the current remote tier state.
-    pub fn refresh_remote_write_mode(&mut self) {
+    /// Swap the read-only sorted map for the configured writable one once the
+    /// current mode says the volume is no longer read-only. Mirrors Go's
+    /// reopenIdxForWrite, and is the reason a volume that stops being read-only
+    /// can accept writes at all: SortedFileNeedleMap::put always fails, so
+    /// publishing a writable volume without this leaves every write appending to
+    /// .dat and then failing to index it. A no-op unless the map is the sorted
+    /// one and the flags have already moved.
+    fn reopen_idx_for_write(&mut self) -> Result<(), VolumeError> {
+        if self.use_sorted_index() || !matches!(self.nm, Some(NeedleMap::SortedFile(_))) {
+            return Ok(());
+        }
+        // load_index only publishes self.nm once the new map is built, so a
+        // failure leaves the sorted map in place and the caller can recover.
+        self.load_index()
+    }
+
+    /// Recompute the Go-style write/delete mode from the current remote tier
+    /// state, and bring the needle map in line with it — a volume that stops
+    /// being remote also stops using the read-only sorted index.
+    ///
+    /// If the map cannot be rebuilt the volume is pinned read-only rather than
+    /// published as writable with an index that rejects every put; a restart
+    /// recovers it.
+    pub fn refresh_remote_write_mode(&mut self) -> Result<(), VolumeError> {
         self.has_remote_file = !self.volume_info.files.is_empty();
         if self.has_remote_file {
             self.no_write_can_delete = true;
@@ -2524,6 +2638,11 @@ impl Volume {
             // only clear the remoteness-derived flag, not an operator mark
             self.no_write_can_delete = false;
         }
+        if let Err(e) = self.reopen_idx_for_write() {
+            self.no_write_or_delete = true;
+            return Err(e);
+        }
+        Ok(())
     }
 
     /// Open the local .dat as the data backend, dropping any remote backend, so reads
@@ -2563,7 +2682,7 @@ impl Volume {
                     }
                 }
                 self.volume_info = pb_info;
-                self.refresh_remote_write_mode();
+                self.refresh_remote_write_mode()?;
                 if self.volume_info.version == 0 {
                     self.volume_info.version = Version::current().0 as u32;
                 }
@@ -2592,7 +2711,7 @@ impl Volume {
                     self.no_write_or_delete = true;
                 }
                 self.volume_info = pb_info;
-                self.refresh_remote_write_mode();
+                self.refresh_remote_write_mode()?;
                 if self.volume_info.version == 0 {
                     self.volume_info.version = Version::current().0 as u32;
                 }
@@ -3165,7 +3284,7 @@ impl Volume {
         // Collect live entries from needle map (sorted ascending)
         let nm = self.nm.as_ref().ok_or(VolumeError::NotInitialized)?;
         let mut entries: Vec<(NeedleId, Offset, Size)> = Vec::new();
-        for (id, nv) in nm.iter_entries() {
+        for (id, nv) in nm.iter_entries().map_err(VolumeError::Io)? {
             if nv.offset.is_zero() || nv.size.is_deleted() {
                 continue;
             }
@@ -3848,6 +3967,29 @@ fn size_mismatch_error(offset: i64, id: NeedleId, found: Size, expected: Size) -
     ))
 }
 
+/// Drop a partial row from the end of `.idx` and return the remaining length.
+///
+/// `.idx` is a run of fixed-size rows, and writers append at EOF: a short write
+/// leaves a partial row behind, and every row appended after it lands off
+/// alignment, so the next load parses the rest of the file as garbage. The
+/// partial row is unrecoverable either way — every loader already skips it — so
+/// removing it before anything can append is what keeps the file parseable.
+/// Go instead refuses to load a volume whose `.idx` is not a whole number of
+/// rows; trimming keeps it mountable, and the rows before the tear are intact.
+fn trim_torn_idx_tail(idx_file: &File, idx_path: &str) -> Result<u64, VolumeError> {
+    let size = idx_file.metadata()?.len();
+    let aligned = size - size % NEEDLE_MAP_ENTRY_SIZE as u64;
+    if aligned != size {
+        warn!(
+            idx = %idx_path,
+            size, aligned, "dropping a partial row from the end of the index"
+        );
+        idx_file.set_len(aligned)?;
+        idx_file.sync_all()?;
+    }
+    Ok(aligned)
+}
+
 pub fn volume_file_name(dir: &str, collection: &str, id: VolumeId) -> String {
     if collection.is_empty() {
         format!("{}/{}", dir, id.0)
@@ -5642,6 +5784,587 @@ mod tests {
             .remove("s3.vif_rw_test");
     }
 
+    /// Build a volume with `needles` needles, tier it to a fake remote backend,
+    /// and reload it the way a restart would — so it comes back read-only with
+    /// its index on disk. The local .dat is left in place, matching the state a
+    /// completed tier-down download leaves behind.
+    fn reload_as_tiered(dir: &str, backend_id: &str, needles: u64) -> Volume {
+        {
+            let mut v = make_test_volume(dir);
+            for i in 1..=needles {
+                let mut n = Needle {
+                    id: NeedleId(i),
+                    cookie: Cookie(0x1234),
+                    data: format!("needle-{i}").into_bytes(),
+                    data_size: format!("needle-{i}").len() as u32,
+                    ..Needle::default()
+                };
+                v.write_needle(&mut n, true, false).unwrap();
+            }
+
+            let vif = VifVolumeInfo {
+                files: vec![VifRemoteFile {
+                    backend_type: "s3".to_string(),
+                    backend_id: backend_id.to_string(),
+                    key: "remote-key".to_string(),
+                    offset: 0,
+                    file_size: v.dat_file_size().unwrap(),
+                    modified_time: 123,
+                    extension: ".dat".to_string(),
+                }],
+                version: v.version().0 as u32,
+                ..VifVolumeInfo::default()
+            };
+            std::fs::write(
+                format!("{}/1.vif", dir),
+                serde_json::to_string_pretty(&vif).unwrap(),
+            )
+            .unwrap();
+
+            let tier_config = crate::remote_storage::s3_tier::S3TierConfig {
+                access_key: "access".to_string(),
+                secret_key: "secret".to_string(),
+                region: "us-east-1".to_string(),
+                bucket: "bucket-a".to_string(),
+                endpoint: "http://127.0.0.1:1".to_string(),
+                storage_class: "STANDARD".to_string(),
+                force_path_style: true,
+            };
+            crate::remote_storage::s3_tier::global_s3_tier_registry()
+                .write()
+                .unwrap()
+                .register(
+                    format!("s3.{backend_id}"),
+                    crate::remote_storage::s3_tier::S3TierBackend::new(&tier_config),
+                );
+        }
+
+        Volume::new(
+            dir,
+            dir,
+            "",
+            VolumeId(1),
+            NeedleMapKind::InMemory,
+            None,
+            None,
+            0,
+            Version::current(),
+        )
+        .unwrap()
+    }
+
+    // Tier-down clears the remote mode and publishes the volume as writable. The
+    // map it booted with is the read-only sorted one, whose put always fails, so
+    // without a rebuild the first write appends to .dat and then cannot be
+    // indexed — leaving bytes nothing references.
+    #[test]
+    fn test_tier_down_swaps_in_a_writable_needle_map() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let mut v = reload_as_tiered(dir, "vif_tierdown_test", 2);
+        assert!(matches!(v.nm, Some(NeedleMap::SortedFile(_))));
+
+        // The tail of the tier-down handler, once the .dat is back on disk.
+        v.volume_info.files.remove(0);
+        v.refresh_remote_write_mode().unwrap();
+        v.save_volume_info().unwrap();
+        v.open_local_dat_backend().unwrap();
+
+        assert!(!v.has_remote_file);
+        assert!(!v.is_read_only(), "tier-down should publish a writable volume");
+        assert!(
+            !matches!(v.nm, Some(NeedleMap::SortedFile(_))),
+            "tier-down must swap the read-only map out before writes are allowed"
+        );
+
+        let mut n = Needle {
+            id: NeedleId(7),
+            cookie: Cookie(0x5678),
+            data: b"after-tier-down".to_vec(),
+            data_size: 15,
+            ..Needle::default()
+        };
+        v.write_needle(&mut n, true, true).unwrap();
+        v.sync_to_disk().unwrap();
+        drop(v);
+
+        // Reload: the needle must come back from .idx, not just have landed in
+        // .dat as unreferenced bytes.
+        let v = Volume::new(
+            dir,
+            dir,
+            "",
+            VolumeId(1),
+            NeedleMapKind::InMemory,
+            None,
+            None,
+            0,
+            Version::current(),
+        )
+        .unwrap();
+        let mut probe = Needle {
+            id: NeedleId(7),
+            ..Needle::default()
+        };
+        v.read_needle(&mut probe).unwrap();
+        assert_eq!(probe.data, b"after-tier-down");
+
+        crate::remote_storage::s3_tier::global_s3_tier_registry()
+            .write()
+            .unwrap()
+            .remove("s3.vif_tierdown_test");
+    }
+
+    // A .sdx that cannot be read end to end must abort compaction. Treating the
+    // short scan as the complete live set would commit a volume missing every
+    // needle past the truncation.
+    #[test]
+    fn test_compaction_aborts_on_unreadable_sorted_index() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let mut v = reload_as_tiered(dir, "vif_compact_test", 4);
+        let Some(NeedleMap::SortedFile(ref nm)) = v.nm else {
+            panic!("tiered volume should search the on-disk .sdx");
+        };
+        let sdx_path = nm.db_file_name().to_string();
+
+        let sdx = OpenOptions::new().write(true).open(&sdx_path).unwrap();
+        sdx.set_len(NEEDLE_MAP_ENTRY_SIZE as u64).unwrap();
+        drop(sdx);
+        crate::storage::needle_map::file_pool::pooled_index_files().discard(&sdx_path);
+
+        let err = v
+            .compact_by_index(0, 0, |_| true)
+            .expect_err("compaction must not commit a partially scanned index");
+        assert!(
+            matches!(err, VolumeError::Io(ref e) if e.kind() == io::ErrorKind::UnexpectedEof),
+            "unexpected error: {err:?}"
+        );
+
+        crate::remote_storage::s3_tier::global_s3_tier_registry()
+            .write()
+            .unwrap()
+            .remove("s3.vif_compact_test");
+    }
+
+    // Building .sdx writes to the index directory, which a read-only volume's
+    // may not allow. Before the sorted map that volume mounted and served reads
+    // off an in-memory index; it must still do so rather than going offline.
+    #[test]
+    #[cfg(unix)]
+    fn test_read_only_volume_mounts_when_the_index_dir_is_not_writable() {
+        use std::os::unix::fs::PermissionsExt;
+
+        // root ignores the directory mode, so there is nothing to simulate.
+        if unsafe { libc::geteuid() } == 0 {
+            return;
+        }
+
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap().to_string();
+        {
+            let mut v = make_test_volume(&dir);
+            let mut n = Needle {
+                id: NeedleId(1),
+                cookie: Cookie(0x1234),
+                data: b"still-readable".to_vec(),
+                data_size: 14,
+                ..Needle::default()
+            };
+            v.write_needle(&mut n, true, false).unwrap();
+            v.set_read_only_persist(false, true).unwrap();
+            v.sync_to_disk().unwrap();
+        }
+
+        let set_mode = |mode: u32| {
+            let mut perms = std::fs::metadata(tmp.path()).unwrap().permissions();
+            perms.set_mode(mode);
+            std::fs::set_permissions(tmp.path(), perms).unwrap();
+        };
+        set_mode(0o555);
+
+        let loaded = Volume::new(
+            &dir,
+            &dir,
+            "",
+            VolumeId(1),
+            NeedleMapKind::InMemory,
+            None,
+            None,
+            0,
+            Version::current(),
+        );
+        set_mode(0o755); // restore before any assertion, so cleanup works
+
+        let v = loaded.expect("a read-only volume must mount without writing its index dir");
+        assert!(
+            !matches!(v.nm, Some(NeedleMap::SortedFile(_))),
+            "the .sdx could not be built, so the index should have fallen back to memory"
+        );
+        let mut probe = Needle {
+            id: NeedleId(1),
+            ..Needle::default()
+        };
+        v.read_needle(&mut probe).unwrap();
+        assert_eq!(probe.data, b"still-readable");
+    }
+
+    // set_writable clears the read-only flags before it can know the .idx writer
+    // will attach. If attaching fails the flags have to go back: a volume that
+    // advertises writable while its map has no writer takes puts into memory and
+    // loses them at the next restart.
+    #[test]
+    #[cfg(unix)]
+    fn test_set_writable_rolls_back_when_the_idx_writer_cannot_attach() {
+        use std::os::unix::fs::PermissionsExt;
+
+        // root ignores the directory mode, so there is nothing to simulate.
+        if unsafe { libc::geteuid() } == 0 {
+            return;
+        }
+
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap().to_string();
+        {
+            let mut v = make_test_volume(&dir);
+            let mut n = Needle {
+                id: NeedleId(1),
+                cookie: Cookie(0x1234),
+                data: b"before".to_vec(),
+                data_size: 6,
+                ..Needle::default()
+            };
+            v.write_needle(&mut n, true, false).unwrap();
+            v.set_read_only_persist(false, true).unwrap();
+            v.sync_to_disk().unwrap();
+        }
+
+        // A read-only mount, both halves of it: nothing new can be created in the
+        // directory, so .sdx generation fails and the index falls back to memory,
+        // and the .idx itself cannot be opened for write, so no writer attaches.
+        let set_mode = |path: &std::path::Path, mode: u32| {
+            let mut perms = std::fs::metadata(path).unwrap().permissions();
+            perms.set_mode(mode);
+            std::fs::set_permissions(path, perms).unwrap();
+        };
+        let idx_path = tmp.path().join("1.idx");
+        set_mode(&idx_path, 0o444);
+        set_mode(tmp.path(), 0o555);
+
+        let loaded = Volume::new(
+            &dir,
+            &dir,
+            "",
+            VolumeId(1),
+            NeedleMapKind::InMemory,
+            None,
+            None,
+            0,
+            Version::current(),
+        );
+        let marked = loaded.map(|mut v| {
+            let r = v.set_writable();
+            (v.no_write_or_delete, v.is_read_only(), v.nm.as_ref().unwrap().has_idx_writer(), r)
+        });
+        // Restore before any assertion, so cleanup works.
+        set_mode(tmp.path(), 0o755);
+        set_mode(&idx_path, 0o644);
+
+        let (no_write_or_delete, read_only, has_writer, result) = marked.unwrap();
+        assert!(
+            result.is_err(),
+            "set_writable cannot succeed without an .idx writer"
+        );
+        assert!(!has_writer, "the writer is what failed to attach");
+        assert!(
+            no_write_or_delete && read_only,
+            "a failed mark must leave the volume read-only, not writable with an index it cannot append to"
+        );
+    }
+
+    // .idx is a run of fixed-size rows and writers append at EOF, so a partial
+    // row left by a short write puts every row appended after it off alignment
+    // and the next load parses the rest as garbage. The tail has to go before
+    // anything writable attaches.
+    #[test]
+    fn test_writable_load_trims_a_torn_idx_tail() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap().to_string();
+        {
+            let mut v = make_test_volume(&dir);
+            let mut n = Needle {
+                id: NeedleId(1),
+                cookie: Cookie(0x1234),
+                data: b"first".to_vec(),
+                data_size: 5,
+                ..Needle::default()
+            };
+            v.write_needle(&mut n, true, true).unwrap();
+            v.sync_to_disk().unwrap();
+        }
+
+        let idx_path = tmp.path().join("1.idx");
+        let whole = std::fs::metadata(&idx_path).unwrap().len();
+        {
+            use std::io::Write as _;
+            let mut f = OpenOptions::new().append(true).open(&idx_path).unwrap();
+            f.write_all(&[0xcd; 5]).unwrap();
+        }
+
+        let mut v = Volume::new(
+            &dir,
+            &dir,
+            "",
+            VolumeId(1),
+            NeedleMapKind::InMemory,
+            None,
+            None,
+            0,
+            Version::current(),
+        )
+        .unwrap();
+        assert_eq!(
+            std::fs::metadata(&idx_path).unwrap().len(),
+            whole,
+            "the partial row should be gone before any writer attaches"
+        );
+
+        let mut n = Needle {
+            id: NeedleId(2),
+            cookie: Cookie(0x5678),
+            data: b"second".to_vec(),
+            data_size: 6,
+            ..Needle::default()
+        };
+        v.write_needle(&mut n, true, true).unwrap();
+        v.sync_to_disk().unwrap();
+        drop(v);
+
+        // Both rows have to survive the round trip; a misaligned append loses
+        // the second one and garbles whatever follows it.
+        let v = Volume::new(
+            &dir,
+            &dir,
+            "",
+            VolumeId(1),
+            NeedleMapKind::InMemory,
+            None,
+            None,
+            0,
+            Version::current(),
+        )
+        .unwrap();
+        for (id, want) in [(1u64, &b"first"[..]), (2u64, &b"second"[..])] {
+            let mut probe = Needle {
+                id: NeedleId(id),
+                ..Needle::default()
+            };
+            v.read_needle(&mut probe).unwrap();
+            assert_eq!(probe.data, want);
+        }
+    }
+
+    // A delete-capable read-only volume — every tiered one — takes the in-memory
+    // loader's read-write branch, which fails on the same unwritable directory
+    // that refused the .sdx. It has to end up read-only rather than offline.
+    #[test]
+    #[cfg(unix)]
+    fn test_delete_capable_volume_falls_all_the_way_back_to_read_only() {
+        use std::os::unix::fs::PermissionsExt;
+
+        // root ignores the directory mode, so there is nothing to simulate.
+        if unsafe { libc::geteuid() } == 0 {
+            return;
+        }
+
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap().to_string();
+        {
+            let mut v = make_test_volume(&dir);
+            let mut n = Needle {
+                id: NeedleId(1),
+                cookie: Cookie(0x1234),
+                data: b"still-readable".to_vec(),
+                data_size: 14,
+                ..Needle::default()
+            };
+            v.write_needle(&mut n, true, false).unwrap();
+            v.set_read_only_persist(true, true).unwrap();
+            v.sync_to_disk().unwrap();
+        }
+
+        let set_mode = |path: &std::path::Path, mode: u32| {
+            let mut perms = std::fs::metadata(path).unwrap().permissions();
+            perms.set_mode(mode);
+            std::fs::set_permissions(path, perms).unwrap();
+        };
+        let idx_path = tmp.path().join("1.idx");
+        set_mode(&idx_path, 0o444);
+        set_mode(tmp.path(), 0o555);
+
+        let loaded = Volume::new(
+            &dir,
+            &dir,
+            "",
+            VolumeId(1),
+            NeedleMapKind::InMemory,
+            None,
+            None,
+            0,
+            Version::current(),
+        );
+        set_mode(tmp.path(), 0o755);
+        set_mode(&idx_path, 0o644);
+
+        let v = loaded.expect("a delete-capable volume must mount without a writable index dir");
+        assert!(
+            v.no_write_or_delete,
+            "with no writer for tombstones, deletes have to be refused outright"
+        );
+        let mut probe = Needle {
+            id: NeedleId(1),
+            ..Needle::default()
+        };
+        v.read_needle(&mut probe).unwrap();
+        assert_eq!(probe.data, b"still-readable");
+    }
+
+    // Issue #10937: a volume server holding hundreds of thousands of cloud-tiered
+    // volumes ran out of descriptors because every one of them pinned its index
+    // for the life of the process. A tiered volume must search the on-disk .sdx
+    // and hold nothing open while it sits idle.
+    #[test]
+    fn test_remote_volume_index_is_on_disk_and_holds_no_descriptors() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+
+        {
+            let mut v = make_test_volume(dir);
+            for i in 1..=4u64 {
+                let mut n = Needle {
+                    id: NeedleId(i),
+                    cookie: Cookie(0x1234),
+                    data: format!("needle-{i}").into_bytes(),
+                    data_size: format!("needle-{i}").len() as u32,
+                    ..Needle::default()
+                };
+                v.write_needle(&mut n, true, false).unwrap();
+            }
+
+            // A writable volume does pin its .idx, which is also what proves the
+            // descriptor probe below is looking at something real.
+            if let Some(count) = crate::storage::needle_map::file_pool::open_index_fds(tmp.path()) {
+                assert!(count > 0, "a writable volume should hold its .idx open");
+            }
+
+            let vif = VifVolumeInfo {
+                files: vec![VifRemoteFile {
+                    backend_type: "s3".to_string(),
+                    backend_id: "vif_fd_test".to_string(),
+                    key: "remote-key".to_string(),
+                    offset: 0,
+                    file_size: v.dat_file_size().unwrap(),
+                    modified_time: 123,
+                    extension: ".dat".to_string(),
+                }],
+                version: v.version().0 as u32,
+                ..VifVolumeInfo::default()
+            };
+            std::fs::write(
+                format!("{}/1.vif", dir),
+                serde_json::to_string_pretty(&vif).unwrap(),
+            )
+            .unwrap();
+
+            let tier_config = crate::remote_storage::s3_tier::S3TierConfig {
+                access_key: "access".to_string(),
+                secret_key: "secret".to_string(),
+                region: "us-east-1".to_string(),
+                bucket: "bucket-a".to_string(),
+                endpoint: "http://127.0.0.1:1".to_string(),
+                storage_class: "STANDARD".to_string(),
+                force_path_style: true,
+            };
+            crate::remote_storage::s3_tier::global_s3_tier_registry()
+                .write()
+                .unwrap()
+                .register(
+                    "s3.vif_fd_test".to_string(),
+                    crate::remote_storage::s3_tier::S3TierBackend::new(&tier_config),
+                );
+        }
+
+        let mut v = Volume::new(
+            dir,
+            dir,
+            "",
+            VolumeId(1),
+            NeedleMapKind::InMemory,
+            None,
+            None,
+            0,
+            Version::current(),
+        )
+        .unwrap();
+
+        assert!(v.has_remote_file);
+        let Some(NeedleMap::SortedFile(ref nm)) = v.nm else {
+            panic!("tiered volume should search the on-disk .sdx, got {:?}", v.nm.is_some());
+        };
+        let idx_path = nm.index_file_name().to_string();
+        let sdx_path = nm.db_file_name().to_string();
+        assert!(Path::new(&sdx_path).exists(), "load should build the .sdx");
+
+        let drop_pooled = || {
+            crate::storage::needle_map::file_pool::pooled_index_files().discard(&idx_path);
+            crate::storage::needle_map::file_pool::pooled_index_files().discard(&sdx_path);
+        };
+
+        if crate::storage::needle_map::file_pool::open_index_fds(tmp.path()).is_some() {
+            drop_pooled();
+            assert_eq!(
+                crate::storage::needle_map::file_pool::open_index_fds(tmp.path()),
+                Some(0),
+                "a loaded tiered volume must hold no .idx/.sdx descriptor"
+            );
+        }
+
+        // Lookups still resolve, straight off .sdx.
+        assert!(!v.nm.as_ref().unwrap().get(NeedleId(3)).unwrap().size.is_deleted());
+        assert!(v.nm.as_ref().unwrap().get(NeedleId(9)).is_none());
+
+        // Deletes are still allowed on a tiered volume and land in .idx.
+        let idx_before = std::fs::metadata(&idx_path).unwrap().len();
+        let deleted = v
+            .delete_needle(&mut Needle {
+                id: NeedleId(3),
+                cookie: Cookie(0x1234),
+                ..Needle::default()
+            })
+            .unwrap();
+        assert!(deleted.0 > 0);
+        assert_eq!(
+            std::fs::metadata(&idx_path).unwrap().len(),
+            idx_before + NEEDLE_MAP_ENTRY_SIZE as u64,
+            "the tombstone must be appended to the .idx tail"
+        );
+        assert!(v.nm.as_ref().unwrap().get(NeedleId(3)).unwrap().size.is_deleted());
+
+        if crate::storage::needle_map::file_pool::open_index_fds(tmp.path()).is_some() {
+            drop_pooled();
+            assert_eq!(
+                crate::storage::needle_map::file_pool::open_index_fds(tmp.path()),
+                Some(0),
+                "reads and deletes must give their borrowed handles back"
+            );
+        }
+
+        crate::remote_storage::s3_tier::global_s3_tier_registry()
+            .write()
+            .unwrap()
+            .remove("s3.vif_fd_test");
+    }
+
     #[test]
     fn test_set_writable_keeps_remote_delete_only_mode() {
         let tmp = TempDir::new().unwrap();
@@ -5657,7 +6380,7 @@ mod tests {
             modified_time: 123,
             extension: ".dat".to_string(),
         });
-        v.refresh_remote_write_mode();
+        v.refresh_remote_write_mode().unwrap();
         v.set_writable().unwrap();
 
         assert!(v.is_read_only());
@@ -5706,12 +6429,16 @@ mod tests {
             "reloaded volume should be read-only from .vif"
         );
         assert!(
-            !v.nm.as_ref().unwrap().has_idx_writer(),
-            "read-only load should not attach an .idx writer"
+            matches!(v.nm, Some(NeedleMap::SortedFile(_))),
+            "read-only load should search the on-disk .sdx"
         );
 
         v.set_writable().unwrap();
         assert!(!v.is_read_only());
+        assert!(
+            !matches!(v.nm, Some(NeedleMap::SortedFile(_))),
+            "set_writable must swap out the read-only map, whose put always fails"
+        );
         assert!(
             v.nm.as_ref().unwrap().has_idx_writer(),
             "set_writable must reattach the .idx writer or post-restart writes vanish"