volume server: refuse to compact a volume tiered to remote storage (#11545)

* volume server: refuse a tier move while compacting, and a commit once tiered

A tier move to remote and a vacuum compaction of the same volume could
interleave and leave the volume unreadable:

- A compaction committing while the upload ran swapped .dat/.idx under
  the transfer, which reopens the .dat by path per part. The move then
  published an object holding the old (or a mixed) layout against the
  compacted .idx, and with keep_local_dat_file=false deleted the only
  compacted .dat.
- A tier move finishing while the compaction copy ran (or between the
  copy and the commit) let the commit swap in the compacted .idx while
  the reload served the pre-compaction remote object through it.

The tier move now refuses to start while the volume is compacting, and
re-checks the compaction revision under the store write lock before it
records the remote file; on a mismatch it deletes the uploaded object
and fails with FailedPrecondition, leaving the volume local. Committing
a compaction on a volume that has a remote file is refused and its
.cpd/.cpx removed, since the reload would read the remote object
through the compacted index.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

* volume server: abort a tier move whose volume was replaced or removed

The tier-up bookkeeping looked the volume up by id only and compared the
compaction revision. A delete and re-create of the same id during the upload
yields a fresh volume at the same revision, so the move recorded the old
volume's object on the new one and, without keep_local_dat_file, removed the
new .dat. An unmounted volume was skipped and the move reported success,
leaving the uploaded object referenced by nothing.

Capture the volume instance (its data-file access control Arc, as the scan
and read plans do) with the revision, and require both under the store write
lock. A replaced volume fails with FailedPrecondition, a missing one with
NotFound; either way nothing is recorded and the object is deleted after the
lock is released. Go fails in both cases because deleting or unmounting closes
the descriptor its copy reads.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

* volume server: refuse to compact a volume tiered to remote storage

Committing a compaction of a tiered volume is refused, since the reload
would read the remote object through the compacted index. The compaction
itself still started: a tiered volume's data backend is the remote object
(the local .dat is dropped or deleted on tier-up), so an explicit vacuum
streamed the whole .dat out of remote storage into a .cpd that the commit
then discarded.

Refuse at the start of the compaction instead, before the .cpd is created,
at the point where Go's copy opens the local .dat. The truncated-index
test now uses a read-only local volume for its sorted index, since a
tiered one no longer reaches the copy.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

* volume server: match the ReadOnly(VolumeId) variant in write_volume_needles

#11543 matched VolumeError::ReadOnly as a unit variant in Store::write_volume_needles, and #11544 changed it to ReadOnly(VolumeId) in the same merge window. Each passed CI on its own, but master no longer compiles the Rust volume server. Carry the volume id through.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>

seaweed-volume/src/server/grpc_server.rs

@@ -7982,6 +7982,68 @@ mod tests {
         assert_eq!(read.unwrap(), b"needle-2");
     }
 
+    // An explicit vacuum of a tiered volume must not copy it out of remote
+    // storage only for the commit to refuse the result.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_compaction_refused_before_copying_a_tiered_volume() {
+        let (service, tmp) = make_local_service_with_volume("", None);
+        seed_compactable_volume(&service);
+        let dat_bytes = std::fs::read(tmp.path().join("1.dat")).unwrap();
+        let (endpoint, shutdown_tx, _deletes) = spawn_fake_s3_server(dat_bytes.clone());
+        register_tier_backend("s3.tiered_then_compact", endpoint);
+        {
+            let mut store = service.state.store.write().unwrap();
+            let (_, vol) = store.find_volume_mut(VolumeId(1)).unwrap();
+            vol.update_remote_files(|files| {
+                files.push(volume_server_pb::RemoteFile {
+                    backend_type: "s3".to_string(),
+                    backend_id: "tiered_then_compact".to_string(),
+                    key: "remote-key".to_string(),
+                    offset: 0,
+                    file_size: dat_bytes.len() as u64,
+                    modified_time: 0,
+                    extension: ".dat".to_string(),
+                })
+            })
+            .unwrap();
+            vol.save_volume_info().unwrap();
+            vol.load_remote_dat_file().unwrap();
+        }
+        std::fs::remove_file(tmp.path().join("1.dat")).unwrap();
+
+        let mut stream = service
+            .vacuum_volume_compact(Request::new(volume_server_pb::VacuumVolumeCompactRequest {
+                volume_id: 1,
+                preallocate: 0,
+            }))
+            .await
+            .unwrap()
+            .into_inner();
+        let mut error = None;
+        while let Some(message) = stream.next().await {
+            if let Err(e) = message {
+                error = Some(e);
+            }
+        }
+        let compacting = {
+            let store = service.state.store.read().unwrap();
+            store.find_volume(VolumeId(1)).unwrap().1.is_compacting()
+        };
+        let read = read_surviving_needle(&service);
+        global_s3_tier_registry()
+            .write()
+            .unwrap()
+            .remove("s3.tiered_then_compact");
+        let _ = shutdown_tx.send(());
+
+        let err = error.expect("a tiered volume must refuse the compaction");
+        assert!(err.message().contains("tiered"), "{err:?}");
+        assert!(!tmp.path().join("1.cpd").exists());
+        assert!(!tmp.path().join("1.cpx").exists());
+        assert!(!compacting);
+        assert_eq!(read.unwrap(), b"needle-2");
+    }
+
     /// Build a local service whose volume has a `.dat` large enough to span
     /// several 2MB copy chunks, so the streaming copy paths are exercised
     /// across multiple messages rather than a single buffer.

seaweed-volume/src/storage/volume.rs

@@ -4408,6 +4408,13 @@ impl Volume {
             return Err(e);
         }
         let idx_size = nm.index_file_size();
+        // The copy would stream the .dat from remote storage for a commit that refuses it.
+        if self.has_remote_file() {
+            return Err(VolumeError::Io(io::Error::other(format!(
+                "volume {} is tiered to remote storage, cannot compact",
+                self.id
+            ))));
+        }
 
         // Fresh opens, not `try_clone`: see `dat_scan_plan`.
         let src_dat = if self.dat_file.is_some() {
@@ -8650,9 +8657,17 @@ mod tests {
     fn test_compaction_aborts_on_truncated_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 mut v = make_test_volume(dir);
+            for i in 1..=4u64 {
+                write_test_needle(&mut v, i, format!("needle-{i}").as_bytes());
+            }
+            v.set_read_only_persist(false, true).unwrap();
+            v.sync_to_disk().unwrap();
+        }
+        let mut v = make_test_volume(dir);
         let Some(NeedleMap::SortedFile(_)) = v.nm else {
-            panic!("tiered volume should search the on-disk .sdx");
+            panic!("read-only volume should search the on-disk .sdx");
         };
 
         let idx = OpenOptions::new()
@@ -8669,11 +8684,6 @@ mod tests {
             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