rust volume: typed errors for store compaction so gRPC can answer NotFound (#11355)

* rust volume: typed errors for store compaction so gRPC can answer NotFound

The vacuum entry points on `Store` returned `Result<_, String>`, so the
gRPC layer had nothing to branch on and answered `Status::internal` for
every failure. A vacuum loop that races a volume being moved or deleted
saw the same code as a disk going bad, and `weed shell` could only tell
the two apart by matching on the message text.

`VolumeError` gains `VolumeNotFound(VolumeId)` — the existing `NotFound`
is needle-level and carries no payload — and `InsufficientSpace`, and
`compact_volume`, `commit_compact_volume`, `cleanup_compact_volume` and
`delete_collection` return it. `impl From<VolumeError> for tonic::Status`
in `server/mod.rs` maps not-found to `not_found`, read-only to
`failed_precondition`, insufficient space to `resource_exhausted`,
already-exists to `already_exists`, and everything else to `internal`;
the four RPCs prefix their own context with `status_with_context`, so a
message reads "commit compact volume 7: volume id 7 is not found". The
store-side "during compact" / "during commit compact" / "during cleaning
up" suffixes are gone, and the free-space message drops the volume id the
prefix already supplies.

`check_compact_volume` had no callers — `VacuumVolumeCheck` computes the
garbage level from its own `find_volume` — and is deleted. `compact_volume`
folded the size estimate into its first lookup, dropping the `unwrap()`
re-lookup that only existed to dodge a borrow.

`ascending_visit` on `CompactNeedleMap`, `RedbNeedleMap`,
`SortedFileNeedleMap` and the `NeedleMap` dispatch is now generic over the
visitor's error type, like `CompactMap::ascending_visit` already was. The
three signatures that can fail on their own bound `E: From<String>` to
carry those failures; the in-memory walk in `iter_entries` names
`Infallible`, which says in the type what its comment used to say in prose.

No Go shell command matches on the old error text: the strings exist only
in weed/storage/store_vacuum.go.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* volume server: trim comments and answer the same codes from Go

- vacuum_volume_check reports VolumeError::VolumeNotFound like the other
  vacuum RPCs instead of its own "not found volume id" wording
- drop doc comments that restate what the code says
- Go volume server wraps ErrVolumeNotFound/ErrInsufficientSpace from
  store_vacuum.go so VacuumVolumeCheck/Compact/Commit/Cleanup and
  DeleteCollection answer NotFound/ResourceExhausted, matching the Rust
  volume server; volumeDeleteStatusError generalized to volumeStatusError

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume server: prefix operation context on vacuum errors

Lower-level errors forwarded by CompactVolume, CommitCompactVolume,
CommitCleanupVolume and DeleteCollection carry no volume id or operation
name. Wrap with %w so the status mapping still sees the sentinel chain,
matching the context the Rust server's status_with_context adds.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume server: map NotEmpty to FailedPrecondition, share mapper in VolumeDelete

Go's volumeStatusError maps ErrVolumeNotEmpty to FailedPrecondition; the
Rust Status conversion was missing it and volume_delete kept a hand-rolled
match. Route it through status_with_context like the vacuum handlers.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

seaweed-volume/src/server/grpc_server.rs

@@ -1125,7 +1125,9 @@ impl VolumeServer for VolumeGrpcService {
         let store = self.state.store.read().unwrap();
         let garbage_ratio = match store.find_volume(vid) {
             Some((_, vol)) => vol.garbage_level(),
-            None => return Err(Status::not_found(format!("not found volume id {}", vid))),
+            None => {
+                return Err(crate::storage::volume::VolumeError::VolumeNotFound(vid).into());
+            }
         };
         Ok(Response::new(volume_server_pb::VacuumVolumeCheckResponse {
             garbage_ratio,
@@ -1183,7 +1185,10 @@ impl VolumeServer for VolumeGrpcService {
                 .inc();
 
             if let Err(e) = result {
-                let _ = tx.blocking_send(Err(Status::internal(e)));
+                let _ = tx.blocking_send(Err(crate::server::status_with_context(
+                    &format!("compact volume {vid}"),
+                    e,
+                )));
             }
         });
 
@@ -1225,7 +1230,10 @@ impl VolumeServer for VolumeGrpcService {
                     volume_size,
                 },
             )),
-            Err(e) => Err(Status::internal(e)),
+            Err(e) => Err(crate::server::status_with_context(
+                &format!("commit compact volume {vid}"),
+                e,
+            )),
         }
     }
 
@@ -1241,7 +1249,10 @@ impl VolumeServer for VolumeGrpcService {
             Ok(()) => Ok(Response::new(
                 volume_server_pb::VacuumVolumeCleanupResponse {},
             )),
-            Err(e) => Err(Status::internal(e)),
+            Err(e) => Err(crate::server::status_with_context(
+                &format!("cleanup volume {vid}"),
+                e,
+            )),
         }
     }
 
@@ -1253,9 +1264,9 @@ impl VolumeServer for VolumeGrpcService {
         let collection = &request.into_inner().collection;
         {
             let mut store = self.state.store.write().unwrap();
-            store
-                .delete_collection(collection)
-                .map_err(Status::internal)?;
+            store.delete_collection(collection).map_err(|e| {
+                crate::server::status_with_context(&format!("delete collection {collection}"), e)
+            })?;
         }
         // The delta the notify path derives is the only thing that tells the
         // master these slots came free: a heartbeat carries the whole list only
@@ -1560,24 +1571,16 @@ impl VolumeServer for VolumeGrpcService {
         let vid = VolumeId(req.volume_id);
         let mut store = self.state.store.write().unwrap();
         if req.only_empty {
-            let (_, vol) = store
-                .find_volume(vid)
-                .ok_or_else(|| Status::not_found(format!("not found volume id {}", vid)))?;
+            let (_, vol) = store.find_volume(vid).ok_or_else(|| {
+                Status::from(crate::storage::volume::VolumeError::VolumeNotFound(vid))
+            })?;
             if vol.file_count() > 0 {
-                return Err(Status::failed_precondition("volume not empty"));
+                return Err(Status::from(crate::storage::volume::VolumeError::NotEmpty));
             }
         }
         store
             .delete_volume(vid, req.only_empty, req.keep_remote_data)
-            .map_err(|e| match e {
-                crate::storage::volume::VolumeError::NotFound => {
-                    Status::not_found(format!("not found volume id {}", vid))
-                }
-                crate::storage::volume::VolumeError::NotEmpty => {
-                    Status::failed_precondition("volume not empty")
-                }
-                other => Status::internal(other.to_string()),
-            })?;
+            .map_err(|e| crate::server::status_with_context(&format!("delete volume {vid}"), e))?;
         self.state.volume_state_notify.notify_one();
         Ok(Response::new(volume_server_pb::VolumeDeleteResponse {}))
     }
@@ -6575,6 +6578,30 @@ mod tests {
         assert_eq!(v.file_count(), 1);
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_vacuum_volume_commit_missing_volume_is_not_found() {
+        let (service, _tmp) = make_local_service_with_volume("vacuum_commit_missing", None);
+
+        let mut request =
+            Request::new(volume_server_pb::VacuumVolumeCommitRequest { volume_id: 4242 });
+        request
+            .extensions_mut()
+            .insert(tonic::transport::server::TcpConnectInfo {
+                local_addr: None,
+                remote_addr: Some("127.0.0.1:65000".parse().unwrap()),
+            });
+
+        let err = service
+            .vacuum_volume_commit(request)
+            .await
+            .expect_err("committing a compaction for a volume that is not mounted must fail");
+        assert_eq!(err.code(), tonic::Code::NotFound, "{err:?}");
+        assert!(
+            err.message().contains("4242"),
+            "the message must still name the volume: {err:?}"
+        );
+    }
+
     #[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/server/mod.rs

@@ -1,3 +1,7 @@
+use tonic::Status;
+
+use crate::storage::volume::VolumeError;
+
 #[cfg(unix)]
 pub mod debug;
 pub mod grpc_client;
@@ -13,3 +17,66 @@ pub mod store_ec;
 pub mod ui;
 pub mod volume_server;
 pub mod write_queue;
+
+/// Map a storage error onto the gRPC code that describes it.
+impl From<VolumeError> for Status {
+    fn from(err: VolumeError) -> Self {
+        let message = err.to_string();
+        match err {
+            VolumeError::NotFound | VolumeError::VolumeNotFound(_) => Status::not_found(message),
+            VolumeError::ReadOnly | VolumeError::NotEmpty => Status::failed_precondition(message),
+            VolumeError::InsufficientSpace { .. } => Status::resource_exhausted(message),
+            VolumeError::AlreadyExists => Status::already_exists(message),
+            _ => Status::internal(message),
+        }
+    }
+}
+
+/// Same mapping, with the RPC's own context prefixed (`compact volume 7: ...`).
+pub fn status_with_context(context: &str, err: VolumeError) -> Status {
+    let status = Status::from(err);
+    Status::new(status.code(), format!("{context}: {}", status.message()))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::storage::types::VolumeId;
+
+    #[test]
+    fn test_volume_error_maps_to_grpc_code() {
+        use tonic::Code;
+
+        let code = |e: VolumeError| Status::from(e).code();
+        assert_eq!(
+            code(VolumeError::VolumeNotFound(VolumeId(7))),
+            Code::NotFound
+        );
+        assert_eq!(code(VolumeError::NotFound), Code::NotFound);
+        assert_eq!(code(VolumeError::ReadOnly), Code::FailedPrecondition);
+        assert_eq!(
+            code(VolumeError::InsufficientSpace {
+                vid: VolumeId(7),
+                required: 2,
+                free: 1,
+            }),
+            Code::ResourceExhausted
+        );
+        assert_eq!(code(VolumeError::AlreadyExists), Code::AlreadyExists);
+        assert_eq!(code(VolumeError::NotInitialized), Code::Internal);
+
+        let status = status_with_context(
+            "compact volume 7",
+            VolumeError::InsufficientSpace {
+                vid: VolumeId(7),
+                required: 2,
+                free: 1,
+            },
+        );
+        assert_eq!(status.code(), Code::ResourceExhausted);
+        assert_eq!(
+            status.message(),
+            "compact volume 7: not enough free space: required 2, free 1"
+        );
+    }
+}

seaweed-volume/src/storage/needle_map.rs

@@ -417,9 +417,9 @@ impl CompactNeedleMap {
     }
 
     /// Visit all entries in ascending order by needle ID.
-    pub fn ascending_visit<F>(&self, f: F) -> Result<(), String>
+    pub fn ascending_visit<F, E>(&self, f: F) -> Result<(), E>
     where
-        F: FnMut(NeedleId, &NeedleValue) -> Result<(), String>,
+        F: FnMut(NeedleId, &NeedleValue) -> Result<(), E>,
     {
         self.map.ascending_visit(f)
     }
@@ -1203,9 +1203,10 @@ impl RedbNeedleMap {
     }
 
     /// Visit all entries in ascending order by needle ID.
-    pub fn ascending_visit<F>(&self, mut f: F) -> Result<(), String>
+    pub fn ascending_visit<F, E>(&self, mut f: F) -> Result<(), E>
     where
-        F: FnMut(NeedleId, &NeedleValue) -> Result<(), String>,
+        F: FnMut(NeedleId, &NeedleValue) -> Result<(), E>,
+        E: From<String>,
     {
         let txn = self
             .db_or_err()
@@ -1443,9 +1444,10 @@ impl NeedleMap {
     }
 
     /// Visit all entries in ascending order by needle ID.
-    pub fn ascending_visit<F>(&self, f: F) -> Result<(), String>
+    pub fn ascending_visit<F, E>(&self, f: F) -> Result<(), E>
     where
-        F: FnMut(NeedleId, &NeedleValue) -> Result<(), String>,
+        F: FnMut(NeedleId, &NeedleValue) -> Result<(), E>,
+        E: From<String>,
     {
         match self {
             NeedleMap::InMemory(nm) => nm.ascending_visit(f),
@@ -1466,7 +1468,7 @@ impl NeedleMap {
                 // The visitor never fails, so neither can this.
                 let _ = nm.ascending_visit(|id, nv| {
                     entries.push((id, *nv));
-                    Ok(())
+                    Ok::<(), std::convert::Infallible>(())
                 });
                 Ok(entries)
             }
@@ -1838,7 +1840,7 @@ mod tests {
         let mut live = 0u64;
         nm.ascending_visit(|_, _| {
             live += 1;
-            Ok(())
+            Ok::<(), String>(())
         })
         .unwrap();
         assert_eq!(live, N - 1);
@@ -2051,7 +2053,7 @@ mod tests {
         let mut visited = Vec::new();
         nm.ascending_visit(|id, nv| {
             visited.push((id, nv.size));
-            Ok(())
+            Ok::<(), String>(())
         })
         .unwrap();
 

seaweed-volume/src/storage/needle_map/sorted_file.rs

@@ -317,9 +317,11 @@ impl SortedFileNeedleMap {
         Ok(())
     }
 
-    pub fn ascending_visit<F>(&self, mut f: F) -> Result<(), String>
+    /// Visit all live entries in ascending order by needle ID.
+    pub fn ascending_visit<F, E>(&self, mut f: F) -> Result<(), E>
     where
-        F: FnMut(NeedleId, &NeedleValue) -> Result<(), String>,
+        F: FnMut(NeedleId, &NeedleValue) -> Result<(), E>,
+        E: From<String>,
     {
         let mut visit_error = None;
         self.visit_live_entries(|id, nv| {
@@ -329,7 +331,7 @@ impl SortedFileNeedleMap {
             }
             Ok(())
         })
-        .map_err(|e| visit_error.take().unwrap_or_else(|| e.to_string()))
+        .map_err(|e| visit_error.take().unwrap_or_else(|| E::from(e.to_string())))
     }
 
     pub fn iter_entries(&self) -> io::Result<Vec<(NeedleId, NeedleValue)>> {
@@ -1035,7 +1037,7 @@ mod tests {
         let mut visited = Vec::new();
         m.ascending_visit(|id, _| {
             visited.push(id);
-            Ok(())
+            Ok::<(), String>(())
         })
         .unwrap();
         assert_eq!(visited, vec![NeedleId(2)]);

seaweed-volume/src/storage/store.rs

@@ -764,10 +764,9 @@ impl Store {
     // ---- Collection operations ----
 
     /// Delete all volumes in a collection.
-    pub fn delete_collection(&mut self, collection: &str) -> Result<(), String> {
+    pub fn delete_collection(&mut self, collection: &str) -> Result<(), VolumeError> {
         for loc in &mut self.locations {
-            loc.delete_collection(collection)
-                .map_err(|e| format!("delete collection {}: {}", collection, e))?;
+            loc.delete_collection(collection)?;
         }
         crate::metrics::delete_collection_metrics(collection);
         Ok(())
@@ -1365,18 +1364,6 @@ impl Store {
 
     // ---- Vacuum / Compaction ----
 
-    /// Check the garbage level of a volume.
-    pub fn check_compact_volume(&self, vid: VolumeId) -> Result<f64, String> {
-        if let Some((_, v)) = self.find_volume(vid) {
-            Ok(v.garbage_level())
-        } else {
-            Err(format!(
-                "volume id {} is not found during check compact",
-                vid.0
-            ))
-        }
-    }
-
     /// Compact a volume by rewriting only live needles.
     pub fn compact_volume<F>(
         &mut self,
@@ -1384,68 +1371,53 @@ impl Store {
         preallocate: u64,
         max_bytes_per_second: i64,
         progress_fn: F,
-    ) -> Result<(), String>
+    ) -> Result<(), VolumeError>
     where
         F: Fn(i64) -> bool,
     {
-        let loc_idx = self
-            .find_volume(vid)
-            .map(|(i, _)| i)
-            .ok_or_else(|| format!("volume id {} is not found during compact", vid.0))?;
+        // Required space matches Go's CompactVolume check: the larger of the
+        // requested preallocation and the estimated volume size.
+        let (loc_idx, space_needed) = {
+            let (loc_idx, v) = self
+                .find_volume(vid)
+                .ok_or(VolumeError::VolumeNotFound(vid))?;
+            let estimated = v.dat_file_size().unwrap_or(0) + v.idx_file_size();
+            (loc_idx, std::cmp::max(preallocate, estimated))
+        };
 
         let dir = self.locations[loc_idx].directory.clone();
         let (_, free) = crate::storage::disk_location::get_disk_stats(&dir);
-
-        // Compute required space: use the larger of preallocate or estimated volume size
-        // matching Go's CompactVolume space check
-        let space_needed = {
-            let (_, v) = self.find_volume(vid).unwrap();
-            let estimated = v.dat_file_size().unwrap_or(0) + v.idx_file_size();
-            std::cmp::max(preallocate, estimated)
-        };
-
         if free < space_needed {
-            return Err(format!(
-                "not enough free space to compact volume {}. Required: {}, Free: {}",
-                vid.0, space_needed, free
-            ));
+            return Err(VolumeError::InsufficientSpace {
+                vid,
+                required: space_needed,
+                free,
+            });
         }
 
-        if let Some((_, v)) = self.find_volume_mut(vid) {
-            v.compact_by_index(preallocate, max_bytes_per_second, progress_fn)
-                .map_err(|e| format!("compact volume {}: {}", vid.0, e))
-        } else {
-            Err(format!("volume id {} is not found during compact", vid.0))
-        }
+        let (_, v) = self
+            .find_volume_mut(vid)
+            .ok_or(VolumeError::VolumeNotFound(vid))?;
+        v.compact_by_index(preallocate, max_bytes_per_second, progress_fn)
     }
 
     /// Commit a completed compaction: swap files and reload.
-    pub fn commit_compact_volume(&mut self, vid: VolumeId) -> Result<(bool, u64), String> {
-        if let Some((_, v)) = self.find_volume_mut(vid) {
-            let is_read_only = v.is_read_only();
-            v.commit_compact()
-                .map_err(|e| format!("commit compact volume {}: {}", vid.0, e))?;
-            let volume_size = v.dat_file_size().unwrap_or(0);
-            Ok((is_read_only, volume_size))
-        } else {
-            Err(format!(
-                "volume id {} is not found during commit compact",
-                vid.0
-            ))
-        }
+    pub fn commit_compact_volume(&mut self, vid: VolumeId) -> Result<(bool, u64), VolumeError> {
+        let (_, v) = self
+            .find_volume_mut(vid)
+            .ok_or(VolumeError::VolumeNotFound(vid))?;
+        let is_read_only = v.is_read_only();
+        v.commit_compact()?;
+        let volume_size = v.dat_file_size().unwrap_or(0);
+        Ok((is_read_only, volume_size))
     }
 
     /// Clean up leftover compaction files.
-    pub fn cleanup_compact_volume(&mut self, vid: VolumeId) -> Result<(), String> {
-        if let Some((_, v)) = self.find_volume_mut(vid) {
-            v.cleanup_compact()
-                .map_err(|e| format!("cleanup volume {}: {}", vid.0, e))
-        } else {
-            Err(format!(
-                "volume id {} is not found during cleaning up",
-                vid.0
-            ))
-        }
+    pub fn cleanup_compact_volume(&mut self, vid: VolumeId) -> Result<(), VolumeError> {
+        let (_, v) = self
+            .find_volume_mut(vid)
+            .ok_or(VolumeError::VolumeNotFound(vid))?;
+        v.cleanup_compact()
     }
 
     /// Close all locations and their volumes.
@@ -2324,6 +2296,88 @@ mod tests {
         assert!(matches!(err, Err(VolumeError::NotFound)));
     }
 
+    #[test]
+    fn test_compaction_of_missing_volume_is_volume_not_found() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let mut store = make_test_store(&[dir]);
+
+        let missing = VolumeId(4242);
+
+        let compact = store.compact_volume(missing, 0, 0, |_| true);
+        assert!(
+            matches!(compact, Err(VolumeError::VolumeNotFound(v)) if v == missing),
+            "{compact:?}"
+        );
+
+        let commit = store.commit_compact_volume(missing);
+        assert!(
+            matches!(commit, Err(VolumeError::VolumeNotFound(v)) if v == missing),
+            "{commit:?}"
+        );
+
+        let cleanup = store.cleanup_compact_volume(missing);
+        assert!(
+            matches!(cleanup, Err(VolumeError::VolumeNotFound(v)) if v == missing),
+            "{cleanup:?}"
+        );
+
+        assert_eq!(
+            VolumeError::VolumeNotFound(missing).to_string(),
+            "volume id 4242 is not found"
+        );
+    }
+
+    #[test]
+    fn test_compact_then_commit_reclaims_a_deleted_needle() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let mut store = make_test_store(&[dir]);
+
+        let vid = VolumeId(1);
+        store
+            .add_volume(vid, DiskType::HardDrive, &VolumeSpec::default())
+            .unwrap();
+
+        for i in 1..=3u64 {
+            let payload = format!("data-{i}").into_bytes();
+            let mut n = Needle {
+                id: NeedleId(i),
+                cookie: Cookie(i as u32),
+                data_size: payload.len() as u32,
+                data: payload,
+                ..Needle::default()
+            };
+            store.write_volume_needle(vid, &mut n, true).unwrap();
+        }
+
+        let mut del = Needle {
+            id: NeedleId(2),
+            cookie: Cookie(2),
+            ..Needle::default()
+        };
+        store.delete_volume_needle(vid, &mut del).unwrap();
+
+        let size_before = store.find_volume(vid).unwrap().1.dat_file_size().unwrap();
+
+        // preallocate 0, unthrottled, progress fn that never cancels.
+        store.compact_volume(vid, 0, 0, |_| true).unwrap();
+
+        let (is_read_only, volume_size) = store.commit_compact_volume(vid).unwrap();
+        assert!(!is_read_only);
+        assert!(
+            volume_size < size_before,
+            "compaction must drop the deleted needle: {volume_size} vs {size_before}"
+        );
+
+        let (_, v) = store
+            .find_volume(vid)
+            .expect("the volume stays mounted across a commit");
+        assert_eq!(v.file_count(), 2);
+        assert_eq!(v.deleted_count(), 0);
+        assert_eq!(v.dat_file_size().unwrap(), volume_size);
+    }
+
     /// Build a Store with N HDD disk locations under a single TempDir.
     /// Returns the store and the TempDir guard so callers keep the dirs
     /// alive for the test's lifetime.

seaweed-volume/src/storage/volume.rs

@@ -38,6 +38,9 @@ pub enum VolumeError {
     #[error("not found")]
     NotFound,
 
+    #[error("volume id {0} is not found")]
+    VolumeNotFound(VolumeId),
+
     #[error("already deleted")]
     Deleted,
 
@@ -62,6 +65,13 @@ pub enum VolumeError {
     #[error("volume size limit exceeded: current {current}, limit {limit}")]
     SizeLimitExceeded { current: u64, limit: u64 },
 
+    #[error("not enough free space: required {required}, free {free}")]
+    InsufficientSpace {
+        vid: VolumeId,
+        required: u64,
+        free: u64,
+    },
+
     #[error("volume not initialized")]
     NotInitialized,
 

weed/server/volume_grpc_admin.go

@@ -82,9 +82,9 @@ func (vs *VolumeServer) DeleteCollection(ctx context.Context, req *volume_server
 
 	if err != nil {
 		glog.Errorf("delete collection %s: %v", req.Collection, err)
-	} else {
-		glog.V(2).Infof("delete collection %v", req)
+		return resp, volumeStatusError(fmt.Errorf("delete collection %s: %w", req.Collection, err))
 	}
+	glog.V(2).Infof("delete collection %v", req)
 
 	return resp, err
 
@@ -202,7 +202,7 @@ func (vs *VolumeServer) VolumeDelete(ctx context.Context, req *volume_server_pb.
 
 	if err != nil {
 		glog.Errorf("volume delete %v: %v", req, err)
-		return resp, volumeDeleteStatusError(err)
+		return resp, volumeStatusError(err)
 	} else {
 		// V(0) so destructive RPCs are always traceable.
 		glog.Infof("volume delete %v", req)
@@ -212,16 +212,19 @@ func (vs *VolumeServer) VolumeDelete(ctx context.Context, req *volume_server_pb.
 
 }
 
-// volumeDeleteStatusError keeps the store's message so callers matching on
+// volumeStatusError keeps the store's message so callers matching on
 // "not found" or "volume not empty" keep working, and adds the status code so
 // new callers do not have to.
-func volumeDeleteStatusError(err error) error {
+func volumeStatusError(err error) error {
 	if errors.Is(err, storage.ErrVolumeNotFound) {
 		return status.Error(codes.NotFound, err.Error())
 	}
 	if errors.Is(err, storage.ErrVolumeNotEmpty) {
 		return status.Error(codes.FailedPrecondition, err.Error())
 	}
+	if errors.Is(err, storage.ErrInsufficientSpace) {
+		return status.Error(codes.ResourceExhausted, err.Error())
+	}
 	return err
 }
 

weed/server/volume_grpc_admin_delete_test.go

@@ -15,18 +15,22 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestVolumeDeleteStatusErrorDistinguishesAbsentFromTransportFailure(t *testing.T) {
-	notFound := volumeDeleteStatusError(fmt.Errorf("delete volume 17 not found on disk: %w", storage.ErrVolumeNotFound))
+func TestVolumeStatusErrorDistinguishesAbsentFromTransportFailure(t *testing.T) {
+	notFound := volumeStatusError(fmt.Errorf("delete volume 17 not found on disk: %w", storage.ErrVolumeNotFound))
 	assert.Equal(t, codes.NotFound, status.Code(notFound))
 	assert.Contains(t, notFound.Error(), "not found", "the store message must survive for callers that match on it")
 
 	transport := errors.New("connection reset")
-	require.ErrorIs(t, volumeDeleteStatusError(transport), transport)
+	require.ErrorIs(t, volumeStatusError(transport), transport)
 	assert.NotEqual(t, codes.NotFound, status.Code(transport))
 
-	notEmpty := volumeDeleteStatusError(storage.ErrVolumeNotEmpty)
+	notEmpty := volumeStatusError(storage.ErrVolumeNotEmpty)
 	assert.Equal(t, codes.FailedPrecondition, status.Code(notEmpty))
 	assert.Contains(t, notEmpty.Error(), "volume not empty")
+
+	noSpace := volumeStatusError(fmt.Errorf("compact volume 17: %w", storage.ErrInsufficientSpace))
+	assert.Equal(t, codes.ResourceExhausted, status.Code(noSpace))
+	assert.Contains(t, noSpace.Error(), "insufficient free space")
 }
 
 func TestVolumeDeleteMapsAbsentStoreVolumeToNotFound(t *testing.T) {

weed/server/volume_grpc_vacuum.go

@@ -2,6 +2,7 @@ package weed_server
 
 import (
 	"context"
+	"fmt"
 	"strconv"
 	"time"
 
@@ -27,9 +28,10 @@ func (vs *VolumeServer) VacuumVolumeCheck(ctx context.Context, req *volume_serve
 
 	if err != nil {
 		glog.V(3).Infof("check volume %d: %v", req.VolumeId, err)
+		return resp, volumeStatusError(err)
 	}
 
-	return resp, err
+	return resp, nil
 
 }
 
@@ -70,7 +72,7 @@ func (vs *VolumeServer) VacuumVolumeCompact(req *volume_server_pb.VacuumVolumeCo
 	stats.VolumeServerVacuumingCompactCounter.WithLabelValues(strconv.FormatBool(err == nil && sendErr == nil)).Inc()
 	if err != nil {
 		glog.Errorf("failed compact volume %d: %v", req.VolumeId, err)
-		return err
+		return volumeStatusError(fmt.Errorf("compact volume %d: %w", req.VolumeId, err))
 	}
 	if sendErr != nil {
 		glog.Errorf("failed compact volume %d report progress: %v", req.VolumeId, sendErr)
@@ -99,15 +101,15 @@ func (vs *VolumeServer) VacuumVolumeCommit(ctx context.Context, req *volume_serv
 
 	readOnly, volumeSize, err := vs.store.CommitCompactVolume(needle.VolumeId(req.VolumeId))
 
-	if err != nil {
-		glog.Errorf("failed commit volume %d: %v", req.VolumeId, err)
-	} else {
-		glog.V(1).Infof("commit volume %d", req.VolumeId)
-	}
 	stats.VolumeServerVacuumingCommitCounter.WithLabelValues(strconv.FormatBool(err == nil)).Inc()
 	resp.IsReadOnly = readOnly
 	resp.VolumeSize = uint64(volumeSize)
-	return resp, err
+	if err != nil {
+		glog.Errorf("failed commit volume %d: %v", req.VolumeId, err)
+		return resp, volumeStatusError(fmt.Errorf("commit compact volume %d: %w", req.VolumeId, err))
+	}
+	glog.V(1).Infof("commit volume %d", req.VolumeId)
+	return resp, nil
 
 }
 
@@ -124,10 +126,10 @@ func (vs *VolumeServer) VacuumVolumeCleanup(ctx context.Context, req *volume_ser
 
 	if err != nil {
 		glog.Errorf("failed cleanup volume %d: %v", req.VolumeId, err)
-	} else {
-		glog.V(1).Infof("cleanup volume %d", req.VolumeId)
+		return resp, volumeStatusError(fmt.Errorf("cleanup volume %d: %w", req.VolumeId, err))
 	}
+	glog.V(1).Infof("cleanup volume %d", req.VolumeId)
 
-	return resp, err
+	return resp, nil
 
 }

weed/storage/store_vacuum.go

@@ -9,12 +9,14 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 )
 
+var ErrInsufficientSpace = fmt.Errorf("insufficient free space")
+
 func (s *Store) CheckCompactVolume(volumeId needle.VolumeId) (float64, error) {
 	if v := s.findVolume(volumeId); v != nil {
 		glog.V(3).Infof("volume %d garbage level: %f", volumeId, v.garbageLevel())
 		return v.garbageLevel(), nil
 	}
-	return 0, fmt.Errorf("volume id %d is not found during check compact", volumeId)
+	return 0, fmt.Errorf("volume id %d is not found during check compact: %w", volumeId, ErrVolumeNotFound)
 }
 
 func (s *Store) CompactVolume(vid needle.VolumeId, preallocate int64, compactionBytePerSecond int64, progressFn ProgressFunc) error {
@@ -28,7 +30,7 @@ func (s *Store) CompactVolume(vid needle.VolumeId, preallocate int64, compaction
 			ProgressCallback:  progressFn,
 		})
 	}
-	return fmt.Errorf("volume id %d is not found during compact", vid)
+	return fmt.Errorf("volume id %d is not found during compact: %w", vid, ErrVolumeNotFound)
 }
 
 func (s *Store) CommitCompactVolume(vid needle.VolumeId) (bool, int64, error) {
@@ -44,14 +46,14 @@ func (s *Store) CommitCompactVolume(vid needle.VolumeId) (bool, int64, error) {
 		}
 		return isReadOnly, volumeSize, err
 	}
-	return false, 0, fmt.Errorf("volume id %d is not found during commit compact", vid)
+	return false, 0, fmt.Errorf("volume id %d is not found during commit compact: %w", vid, ErrVolumeNotFound)
 }
 
 func (s *Store) CommitCleanupVolume(vid needle.VolumeId) error {
 	if v := s.findVolume(vid); v != nil {
 		return v.cleanupCompact()
 	}
-	return fmt.Errorf("volume id %d is not found during cleaning up", vid)
+	return fmt.Errorf("volume id %d is not found during cleaning up: %w", vid, ErrVolumeNotFound)
 }
 
 func ensureCompactVolumeSpace(v *Volume, preallocate int64) error {
@@ -69,8 +71,8 @@ func ensureCompactVolumeSpace(v *Volume, preallocate int64) error {
 
 	diskStatus := stats.NewDiskStatus(v.dir)
 	if int64(diskStatus.Free) < spaceNeeded {
-		return fmt.Errorf("insufficient free space for compaction: need %d bytes (volume: %d, index: %d), but only %d bytes available",
-			spaceNeeded, volumeSize, indexSize, diskStatus.Free)
+		return fmt.Errorf("insufficient free space for compaction: need %d bytes (volume: %d, index: %d), but only %d bytes available: %w",
+			spaceNeeded, volumeSize, indexSize, diskStatus.Free, ErrInsufficientSpace)
 	}
 
 	glog.V(1).Infof("volume %d compaction space check: volume=%d, index=%d, space_needed=%d, free_space=%d",