rust volume: mirror the VolumeConsolidateIndex RPC from Go (#10752)

The Go volume server has VolumeConsolidateIndex, which moves a volume's
.idx out of the data directory into the configured -dir.idx directory
(where an EC decode/reconstruct can leave it co-located) and reloads the
volume in place. The Rust port's proto omitted the RPC entirely, so its
generated VolumeServer trait was one method short of Go's.

Add the proto message and rpc, the gated grpc handler, and
Store::consolidate_volume_index / Volume::relocate_index_to, mirroring
Go's Store.ConsolidateVolumeIndex and Volume.RelocateIndexTo -- including
the cross-device copy fallback and the reopen-against-the-old-dir path
when the move fails.

Integration tests cover the real move (index relocated, volume still
serves reads and the move is idempotent), the no-op paths (index already
in place, no separate idx dir) and the not-found error, plus the grpc
handler end to end.

seaweed-volume/proto/volume_server.proto

@@ -45,6 +45,8 @@ service VolumeServer {
     }
     rpc VolumeUnmount (VolumeUnmountRequest) returns (VolumeUnmountResponse) {
     }
+    rpc VolumeConsolidateIndex (VolumeConsolidateIndexRequest) returns (VolumeConsolidateIndexResponse) {
+    }
     rpc VolumeDelete (VolumeDeleteRequest) returns (VolumeDeleteResponse) {
     }
     rpc VolumeMarkReadonly (VolumeMarkReadonlyRequest) returns (VolumeMarkReadonlyResponse) {
@@ -244,6 +246,12 @@ message VolumeUnmountRequest {
 message VolumeUnmountResponse {
 }
 
+message VolumeConsolidateIndexRequest {
+    uint32 volume_id = 1;
+}
+message VolumeConsolidateIndexResponse {
+}
+
 message VolumeDeleteRequest {
     uint32 volume_id = 1;
     bool only_empty = 2;

seaweed-volume/src/server/grpc_server.rs

@@ -952,6 +952,22 @@ impl VolumeServer for VolumeGrpcService {
         Ok(Response::new(volume_server_pb::VolumeUnmountResponse {}))
     }
 
+    async fn volume_consolidate_index(
+        &self,
+        request: Request<volume_server_pb::VolumeConsolidateIndexRequest>,
+    ) -> Result<Response<volume_server_pb::VolumeConsolidateIndexResponse>, Status> {
+        self.check_grpc_admin_auth(&request)?;
+        self.state.check_maintenance()?;
+        let vid = VolumeId(request.into_inner().volume_id);
+        let mut store = self.state.store.write().unwrap();
+        store
+            .consolidate_volume_index(vid)
+            .map_err(|e| Status::internal(e.to_string()))?;
+        Ok(Response::new(
+            volume_server_pb::VolumeConsolidateIndexResponse {},
+        ))
+    }
+
     async fn volume_delete(
         &self,
         request: Request<volume_server_pb::VolumeDeleteRequest>,
@@ -5416,6 +5432,29 @@ mod tests {
         (VolumeGrpcService { state }, tmp)
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_volume_consolidate_index_rpc() {
+        let (service, _tmp) = make_local_service_with_volume("consolidate_rpc", None);
+
+        // Carry a peer address so the admin gate sees a caller; the test guard
+        // has an empty whitelist, so any peer is accepted.
+        let mut request = Request::new(volume_server_pb::VolumeConsolidateIndexRequest {
+            volume_id: 1,
+        });
+        request.extensions_mut().insert(tonic::transport::server::TcpConnectInfo {
+            local_addr: None,
+            remote_addr: Some("127.0.0.1:65000".parse().unwrap()),
+        });
+
+        service.volume_consolidate_index(request).await.unwrap();
+
+        // Data and index share a directory here, so there is nothing to move;
+        // the volume stays mounted and keeps its needle.
+        let store = service.state.store.read().unwrap();
+        let (_, v) = store.find_volume(VolumeId(1)).unwrap();
+        assert_eq!(v.file_count(), 1);
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn test_volume_incremental_copy_streams_remote_only_volume_data() {
         let (service, _tmp, shutdown_tx, dat_bytes, super_block_size, _delete_count) =

seaweed-volume/src/storage/store.rs

@@ -156,6 +156,26 @@ impl Store {
         self.find_volume(vid).is_some()
     }
 
+    /// Move a volume's index into the configured `-dir.idx` directory, reloading
+    /// the volume in place. A no-op when the location has no separate index
+    /// directory. Mirrors Go's `Store::ConsolidateVolumeIndex`.
+    pub fn consolidate_volume_index(&mut self, vid: VolumeId) -> Result<(), VolumeError> {
+        for loc in self.locations.iter_mut() {
+            let idx_dir = loc.idx_directory.clone();
+            let data_dir = loc.directory.clone();
+            if let Some(v) = loc.find_volume_mut(vid) {
+                if idx_dir == data_dir {
+                    return Ok(());
+                }
+                return v.relocate_index_to(&idx_dir);
+            }
+        }
+        Err(VolumeError::Io(io::Error::new(
+            io::ErrorKind::NotFound,
+            format!("volume {} not found on disk", vid),
+        )))
+    }
+
     // ---- Volume lifecycle ----
 
     /// Find the location with fewest volumes (load-balance) of the given disk type.
@@ -1388,6 +1408,73 @@ mod tests {
         assert!(deleted.0 > 0);
     }
 
+    #[test]
+    fn test_consolidate_volume_index_not_found() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let mut store = make_test_store(&[dir]);
+        let err = store.consolidate_volume_index(VolumeId(9)).unwrap_err();
+        assert!(matches!(err, VolumeError::Io(ref e)
+            if e.kind() == std::io::ErrorKind::NotFound));
+    }
+
+    #[test]
+    fn test_consolidate_volume_index_noop_with_separate_idx_dir() {
+        let root = TempDir::new().unwrap();
+        let data_dir = root.path().join("data");
+        let idx_dir = root.path().join("idx");
+        std::fs::create_dir_all(&data_dir).unwrap();
+        std::fs::create_dir_all(&idx_dir).unwrap();
+
+        let mut store = Store::new(NeedleMapKind::InMemory);
+        store
+            .add_location(
+                data_dir.to_str().unwrap(),
+                idx_dir.to_str().unwrap(),
+                10,
+                DiskType::HardDrive,
+                MinFreeSpace::Percent(1.0),
+                Vec::new(),
+            )
+            .unwrap();
+        store
+            .add_volume(
+                VolumeId(1),
+                "",
+                None,
+                None,
+                0,
+                DiskType::HardDrive,
+                Version::current(),
+            )
+            .unwrap();
+        let mut n = Needle {
+            id: NeedleId(1),
+            cookie: Cookie(0xaa),
+            data: b"co-located".to_vec(),
+            data_size: 10,
+            ..Needle::default()
+        };
+        store.write_volume_needle(VolumeId(1), &mut n).unwrap();
+
+        // add_volume already placed the index in the -dir.idx directory, so
+        // consolidation has nothing to move and leaves the volume readable.
+        store.consolidate_volume_index(VolumeId(1)).unwrap();
+        let idx_file = format!(
+            "{}.idx",
+            volume_file_name(idx_dir.to_str().unwrap(), "", VolumeId(1))
+        );
+        assert!(std::path::Path::new(&idx_file).exists());
+
+        let mut got = Needle {
+            id: NeedleId(1),
+            ..Needle::default()
+        };
+        let count = store.read_volume_needle(VolumeId(1), &mut got).unwrap();
+        assert_eq!(count, 10);
+        assert_eq!(got.data, b"co-located");
+    }
+
     #[test]
     fn test_store_multi_location() {
         let tmp1 = TempDir::new().unwrap();

seaweed-volume/src/storage/volume.rs

@@ -17,7 +17,7 @@ use std::sync::Arc;
 use std::sync::{Condvar, Mutex};
 use std::time::{SystemTime, UNIX_EPOCH};
 
-use tracing::{info, warn};
+use tracing::{error, info, warn};
 
 #[cfg(test)]
 use crate::storage::idx;
@@ -3512,6 +3512,62 @@ impl Volume {
         self.nm = None;
     }
 
+    /// Move this volume's `.idx` (and `.sdx` when present) from its current
+    /// index directory into `new_idx_dir`, then reload in place. Mirrors Go's
+    /// `Volume::RelocateIndexTo`: it is a no-op when the index is already there
+    /// or nothing is co-located to move, and is used to pull an index a decode
+    /// or reconstruct left beside the data back into the configured `-dir.idx`.
+    pub fn relocate_index_to(&mut self, new_idx_dir: &str) -> Result<(), VolumeError> {
+        let _guard = self.data_file_access_control.write_lock();
+
+        if self.dir_idx == new_idx_dir {
+            return Ok(());
+        }
+        let old_base = volume_file_name(&self.dir_idx, &self.collection, self.id);
+        let old_idx = format!("{old_base}.idx");
+        if !Path::new(&old_idx).exists() {
+            return Ok(()); // nothing co-located to move
+        }
+        let new_base = volume_file_name(new_idx_dir, &self.collection, self.id);
+        let new_idx = format!("{new_base}.idx");
+
+        // Close the index and data handles before moving the index file.
+        let version = self.version();
+        self.close();
+
+        if let Err(e) = rename_or_copy(&old_idx, &new_idx) {
+            // Reopen against the old dir so the volume is not left down; surface
+            // a failed reopen since it leaves the volume unusable until reload.
+            if let Err(reopen) = self.load(true, false, 0, version) {
+                error!(
+                    "relocate volume {}: reopen after failed .idx move: {}",
+                    self.id, reopen
+                );
+            }
+            return Err(VolumeError::Io(io::Error::new(
+                io::ErrorKind::Other,
+                format!("relocate index for volume {}: move .idx: {e}", self.id),
+            )));
+        }
+
+        // The .sdx is a derived sorted index; move it when present, but a
+        // failure is not fatal — drop the stale copy so a reload rebuilds it.
+        let old_sdx = format!("{old_base}.sdx");
+        if Path::new(&old_sdx).exists() {
+            let new_sdx = format!("{new_base}.sdx");
+            if let Err(e) = rename_or_copy(&old_sdx, &new_sdx) {
+                warn!(
+                    "relocate volume {}: move .sdx: {} (will rebuild)",
+                    self.id, e
+                );
+                let _ = fs::remove_file(&old_sdx);
+            }
+        }
+
+        self.dir_idx = new_idx_dir.to_string();
+        self.load(true, false, 0, version)
+    }
+
     /// Remove all volume files from disk.
     /// Destroy removes everything related to this volume. When keep_remote_data
     /// is true the cloud-tier object backing the volume is left intact — used
@@ -3654,6 +3710,34 @@ pub fn volume_file_name(dir: &str, collection: &str, id: VolumeId) -> String {
     }
 }
 
+/// Move `src` to `dst`, falling back to copy+remove when the two live on
+/// different filesystems (a real `-dir.idx` split). Mirrors Go's
+/// `RenameOrCopyFile`: any rename error other than cross-device is fatal, and a
+/// failed copy or a source that cannot be removed rolls the copy back so a
+/// failure never leaves two divergent files.
+fn rename_or_copy(src: &str, dst: &str) -> io::Result<()> {
+    match fs::rename(src, dst) {
+        Ok(()) => return Ok(()),
+        Err(e) if e.raw_os_error() != Some(libc::EXDEV) => return Err(e),
+        Err(_) => {}
+    }
+    fs::copy(src, dst)?;
+    // fsync the destination before dropping the source, matching Go's out.Sync().
+    if let Err(e) = OpenOptions::new()
+        .write(true)
+        .open(dst)
+        .and_then(|f| f.sync_all())
+    {
+        let _ = fs::remove_file(dst);
+        return Err(e);
+    }
+    if let Err(e) = fs::remove_file(src) {
+        let _ = fs::remove_file(dst);
+        return Err(e);
+    }
+    Ok(())
+}
+
 /// Generate a monotonically increasing append timestamp.
 fn get_append_at_ns(last: u64) -> u64 {
     let now = SystemTime::now()
@@ -4535,6 +4619,96 @@ mod tests {
         assert_eq!(std::str::from_utf8(&n.data).unwrap(), "data 2");
     }
 
+    #[test]
+    fn test_relocate_index_to_moves_index_and_serves_reads() {
+        let root = TempDir::new().unwrap();
+        let data_dir = root.path().join("data");
+        let idx_dir = root.path().join("idx");
+        fs::create_dir_all(&data_dir).unwrap();
+        fs::create_dir_all(&idx_dir).unwrap();
+        let data = data_dir.to_str().unwrap();
+        let idx = idx_dir.to_str().unwrap();
+
+        // Create the volume with its index co-located in the data dir (the state
+        // an EC reconstruct leaves), write a needle, and flush.
+        let mut v = Volume::new(
+            data,
+            data,
+            "",
+            VolumeId(7),
+            NeedleMapKind::InMemory,
+            None,
+            None,
+            0,
+            Version::current(),
+        )
+        .unwrap();
+        let payload = b"payload-across-relocate".to_vec();
+        let mut n = Needle {
+            id: NeedleId(42),
+            cookie: Cookie(0x55),
+            data: payload.clone(),
+            data_size: payload.len() as u32,
+            ..Needle::default()
+        };
+        v.write_needle(&mut n, true).unwrap();
+        v.sync_to_disk().unwrap();
+
+        let data_idx = format!("{data}/7.idx");
+        let idx_dir_idx = format!("{idx}/7.idx");
+        assert!(
+            Path::new(&data_idx).exists(),
+            "precondition: index co-located with the data"
+        );
+
+        v.relocate_index_to(idx).unwrap();
+
+        assert!(Path::new(&idx_dir_idx).exists(), "index moved to the idx dir");
+        assert!(
+            !Path::new(&data_idx).exists(),
+            "index gone from the data dir"
+        );
+        assert_eq!(v.file_count(), 1);
+
+        // The volume reloaded in place, so the needle still reads back.
+        let mut got = Needle {
+            id: NeedleId(42),
+            ..Needle::default()
+        };
+        v.read_needle(&mut got).unwrap();
+        assert_eq!(got.data, payload, "needle content survives the relocate");
+
+        // Relocating again is a no-op: the index is already where asked.
+        v.relocate_index_to(idx).unwrap();
+        assert!(Path::new(&idx_dir_idx).exists());
+    }
+
+    #[test]
+    fn test_relocate_index_to_noop_when_already_in_place() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let mut v = make_test_volume(dir);
+        let mut n = Needle {
+            id: NeedleId(1),
+            cookie: Cookie(1),
+            data: b"x".to_vec(),
+            data_size: 1,
+            ..Needle::default()
+        };
+        v.write_needle(&mut n, true).unwrap();
+        v.sync_to_disk().unwrap();
+
+        // Data and index share a directory, so there is nothing to move.
+        v.relocate_index_to(dir).unwrap();
+        assert!(Path::new(&format!("{dir}/1.idx")).exists());
+        let mut got = Needle {
+            id: NeedleId(1),
+            ..Needle::default()
+        };
+        v.read_needle(&mut got).unwrap();
+        assert_eq!(got.data, b"x");
+    }
+
     #[test]
     fn test_volume_cookie_mismatch() {
         let tmp = TempDir::new().unwrap();