fix(volume [rust]): compare live compaction_revision instead of stale last_compact_revision (#10189)

* fix(volume [rust]): compare live compaction_revision instead of stale last_compact_revision

* fix(volume [rust]): compare live compaction_revision instead of stale last_compact_revision - unit tests

* s3: invalidate stale reader cache locations on chunk read failure (#10156)

* s3: invalidate stale reader cache locations on chunk read failure

* filer: share the chunk-read self-heal across reader cache and streaming paths

The reader cache retry added a third copy of the invalidate-relookup-compare-retry
dance already inlined in PrepareStreamContentWithThrottler and duplicated in
retryWithCacheInvalidation. Extract retryFetchWithFreshLocations and route all
three through it, parameterized by the refetch primitive.

* filer: drop redundant completedTimeNew store in reader cache success path

startCaching already stamps completedTimeNew unconditionally before the
fetchErr branch; the second store inside the success branch is dead.

* filer: make NewReaderCache cache invalidator an explicit parameter

The variadic ...CacheInvalidator only ever read the first element, so a caller
could pass two and silently get one. Take a single explicit argument and have
the non-S3 callers pass nil.

* filer: inject reader cache chunk fetch as a struct field

Replace the process-global readerCacheFetchChunkData test seam with a
per-instance fetchChunkDataFn field defaulted in NewReaderCache, matching how
lookupFileIdFn is already wired. Tests set the field on the cache instead of
swapping a shared global.

* filer: log the location count, not full URLs, on self-heal retry

---------

Co-authored-by: Chris Lu <chris.lu@gmail.com>

* fix(shell): honor explicit fs.mergeVolumes from/to direction (#10159)

* fix(shell): honor explicit fs.mergeVolumes from/to direction

mergeVolumes only ever merged a smaller volume into a larger one. When the
user named both -fromVolumeId and -toVolumeId with the source larger than the
target, the planner produced an empty plan and the command printed just
"max volume size: N MB" and moved nothing.

Build the requested pair directly when both ids are given, instead of routing
through the size-descending heuristic. Read-only, empty, and wrong-collection
endpoints are rejected with a clear error rather than a silent no-op.

* fix(shell): allow fs.mergeVolumes into an empty target volume

Merging chunks into an empty volume is valid, e.g. consolidating data into a
freshly created or recently vacuumed volume. Only reject an empty source, which
has nothing to move.

* fix(shell): reject self-map in directed mergeVolumes planner

createMergePlan with from == to returned a {vid: vid} self-merge when called
directly. Guard it in the planner so it is correct independent of the Do
entrypoint.

* fix(volume [rust]): compare compaction_revision in u32, not truncated u16

`req.compaction_revision as u16` truncates any request value above 65535, so a
stale revision of 65537 aliases to a live revision of 1 and the "is compacted"
guard wrongly passes. Widen the volume's revision to u32 and compare there,
matching Go's uint32(v.CompactionRevision) != req.CompactionRevision.

---------

Co-authored-by: adri <adri@digitalunited.net>
Co-authored-by: Aleksey <48918167+MilanFun@users.noreply.github.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

@@ -1516,9 +1516,16 @@ impl VolumeServer for VolumeGrpcService {
                 .find_volume(vid)
                 .ok_or_else(|| Status::not_found(format!("not found volume id {}", vid)))?;
 
-            // Check compaction revision
+            // Check compaction revision. Compare against the volume's live
+            // super_block.compaction_revision (matching Go's v.CompactionRevision,
+            // which is the embedded SuperBlock field). last_compact_revision() is
+            // a separate bookkeeping value recorded just before a compaction starts
+            // (for makeup-diff catch-up) and is intentionally left behind the live
+            // revision afterward — it is not interchangeable with the live value and
+            // must not be used here, or this check spuriously fails on every volume
+            // that has ever been compacted even once.
             if req.compaction_revision != u32::MAX
-                && v.last_compact_revision() != req.compaction_revision as u16
+                && u32::from(v.super_block.compaction_revision) != req.compaction_revision
             {
                 return Err(Status::failed_precondition(format!(
                     "volume {} is compacted",
@@ -5658,6 +5665,105 @@ mod tests {
         }
     }
 
+    // Regression test for comparing the wrong compaction-revision field.
+    // last_compact_revision() is bookkeeping recorded just before a compaction
+    // starts (for makeup-diff catch-up) and is intentionally left behind
+    // super_block.compaction_revision once the compaction commits. copy_file's
+    // precondition check must compare against the live super_block value
+    // (matching Go's v.CompactionRevision), not the stale bookkeeping one, or
+    // it fails "volume N is compacted" on every volume that has ever been
+    // compacted even once.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn test_copy_file_accepts_live_revision_after_compaction() {
+        let (service, _tmp) = make_local_service_with_volume("", None);
+
+        // Drive a real compaction to completion so the two revision fields
+        // actually diverge the way they do in production, rather than poking
+        // the struct fields directly.
+        let (revision_before, revision_after, last_compact_revision_after) = {
+            let mut store = service.state.store.write().unwrap();
+            let (_, v) = store.find_volume_mut(VolumeId(1)).unwrap();
+            let before = v.super_block.compaction_revision;
+
+            v.compact_by_index(0, 0, |_| true).unwrap();
+            v.commit_compact().unwrap();
+
+            (before, v.super_block.compaction_revision, v.last_compact_revision())
+        };
+
+        assert_eq!(revision_before, 0, "fresh volume starts at revision 0");
+        assert_eq!(
+            revision_after, 1,
+            "live super_block.compaction_revision must advance after commit_compact"
+        );
+        assert_eq!(
+            last_compact_revision_after, 0,
+            "last_compact_revision() is recorded pre-compaction and must stay \
+             behind the live revision — this divergence is exactly what made \
+             the old check fail permanently"
+        );
+
+        // The live revision (1) must be accepted: this is what
+        // read_volume_file_status would report to a caller right now.
+        let response = service
+            .copy_file(Request::new(volume_server_pb::CopyFileRequest {
+                volume_id: 1,
+                ext: ".dat".to_string(),
+                compaction_revision: revision_after as u32,
+                stop_offset: u64::MAX,
+                collection: String::new(),
+                is_ec_volume: false,
+                ignore_source_file_not_found: false,
+            }))
+            .await;
+        assert!(
+            response.is_ok(),
+            "copy_file must accept the live compaction_revision, got: {:?}",
+            response.err()
+        );
+
+        // A genuinely stale/wrong revision must still be rejected — the fix
+        // corrects which field is compared, it does not disable the check.
+        let stale_response = service
+            .copy_file(Request::new(volume_server_pb::CopyFileRequest {
+                volume_id: 1,
+                ext: ".dat".to_string(),
+                compaction_revision: 99,
+                stop_offset: u64::MAX,
+                collection: String::new(),
+                is_ec_volume: false,
+                ignore_source_file_not_found: false,
+            }))
+            .await;
+        let err = match stale_response {
+            Ok(_) => panic!("a genuinely stale revision must be rejected"),
+            Err(e) => e,
+        };
+        assert_eq!(err.code(), tonic::Code::FailedPrecondition);
+        assert!(err.message().contains("is compacted"));
+
+        // A request revision that only collides after truncating to u16 must be
+        // rejected: 65537 as u16 == 1 == the live revision, but as u32 they differ.
+        // Compare in u32 space (matching Go) so this cannot spuriously pass.
+        let truncating_response = service
+            .copy_file(Request::new(volume_server_pb::CopyFileRequest {
+                volume_id: 1,
+                ext: ".dat".to_string(),
+                compaction_revision: u32::from(revision_after) + (1 << 16),
+                stop_offset: u64::MAX,
+                collection: String::new(),
+                is_ec_volume: false,
+                ignore_source_file_not_found: false,
+            }))
+            .await;
+        let err = match truncating_response {
+            Ok(_) => panic!("a revision that only matches after u16 truncation must be rejected"),
+            Err(e) => e,
+        };
+        assert_eq!(err.code(), tonic::Code::FailedPrecondition);
+        assert!(err.message().contains("is compacted"));
+    }
+
     #[tokio::test]
     async fn test_volume_ec_shards_generate_persists_expire_at_sec() {
         let ttl = crate::storage::needle::ttl::TTL::read("3m").unwrap();

seaweed-volume/src/server/heartbeat.rs

@@ -886,7 +886,7 @@ fn build_heartbeat_with_ec_status(
                     replica_placement: vol.super_block.replica_placement.to_byte() as u32,
                     version: vol.super_block.version.0 as u32,
                     ttl: vol.super_block.ttl.to_u32(),
-                    compact_revision: vol.last_compact_revision() as u32,
+                    compact_revision: vol.super_block.compaction_revision as u32,
                     modified_at_second: vol.last_modified_ts() as i64,
                     disk_type: loc.disk_type.to_string(),
                     disk_id: disk_id as u32,