From cc4043c9d22bda855a63f212a851d350c2c9e0f2 Mon Sep 17 00:00:00 2001 From: adri <74632179+justadri@users.noreply.github.com> Date: Thu, 2 Jul 2026 00:36:44 -0400 Subject: [PATCH] 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 * 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 Co-authored-by: Aleksey <48918167+MilanFun@users.noreply.github.com> Co-authored-by: Chris Lu Co-authored-by: Chris Lu --- seaweed-volume/src/server/grpc_server.rs | 110 ++++++++++++++++++++++- seaweed-volume/src/server/heartbeat.rs | 2 +- 2 files changed, 109 insertions(+), 3 deletions(-) diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 1371a4820..2b05765c6 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/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(); diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index 1601d7efb..c0fcb3cdc 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/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,