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"