diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index e04707eba..1034d0a06 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -94,6 +94,12 @@ fn is_skippable_needle_read_error(e: &VolumeError) -> bool { } } +/// Reports whether the compacted .dat is short of the live bytes the +/// pre-compaction index snapshot expected. +fn exceeds_expected_compacted_size(expected_live_bytes: u64, dst_dat_size: u64) -> bool { + expected_live_bytes > dst_dat_size +} + /// Returns true for I/O errors that indicate faulty storage media, not /// transient/network failures. On Unix this is EIO; on Windows it covers /// ERROR_CRC and ERROR_IO_DEVICE, which the kernel returns for failing disks. @@ -3592,6 +3598,7 @@ impl Volume { let mut skipped_needles: u64 = 0; let mut skipped_data_bytes: u64 = 0; + let mut expected_live_bytes: u64 = 0; for (id, offset, size) in entries { // Progress callback if !progress_fn(offset.to_actual_offset()) { @@ -3655,6 +3662,11 @@ impl Volume { } } + // Tally the live bytes from the frozen snapshot this loop copied + // from, not the live needle map. Unreadable needles return before + // this point, so no further skipped-byte adjustment is needed. + expected_live_bytes += size.0 as u64; + // Write needle to destination let bytes = n.write_bytes(version); dst.write_all(&bytes)?; @@ -3675,6 +3687,21 @@ impl Volume { dst.sync_all()?; + if self.super_block.ttl.is_empty() { + let dst_dat_size = dst.metadata()?.len(); + if exceeds_expected_compacted_size(expected_live_bytes, dst_dat_size) { + let _ = fs::remove_file(&cpd_path); + let _ = fs::remove_file(&cpx_path); + return Err(VolumeError::Io(io::Error::new( + io::ErrorKind::UnexpectedEof, + format!( + "volume {} unexpected new data size: {} does not match expected live content size {} from the pre-compaction snapshot", + self.id.0, dst_dat_size, expected_live_bytes + ), + ))); + } + } + // Save new index new_nm.save_to_idx(&cpx_path)?; @@ -6288,6 +6315,64 @@ mod tests { v.cleanup_compact().unwrap(); } + /// Guards the copy-phase integrity check against regressing into + /// double-subtracting skipped bytes: expected_live_bytes already excludes + /// needles dropped as unreadable (they continue before the tally), so the + /// check must compare it directly against the compacted .dat size. + #[test] + fn test_exceeds_expected_compacted_size() { + assert!(!exceeds_expected_compacted_size(100, 100)); + assert!(!exceeds_expected_compacted_size(100, 150)); + assert!(exceeds_expected_compacted_size(100, 90)); + } + + /// A write that lands on the live volume mid-copy must not trip the + /// post-copy integrity check. The Rust port snapshots the index entries + /// before the copy loop (equivalent to Go's frozen oldNm), so a concurrent + /// write is invisible to the tally and the check stays quiet. The write + /// is then replayed by makeup_diff during commit_compact. + #[test] + fn test_compact_by_index_tolerates_concurrent_write() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut v = make_test_volume(dir); + + for i in 1..=8u64 { + let mut n = Needle { + id: NeedleId(i), + cookie: Cookie(i as u32), + data: format!("data-{}", i).into_bytes(), + data_size: format!("data-{}", i).len() as u32, + ..Needle::default() + }; + v.write_needle(&mut n, true, false).unwrap(); + } + + v.compact_by_index(0, 0, |_| true).unwrap(); + assert!(Path::new(&v.file_name(".cpd")).exists()); + + // A write arriving after the snapshot but before commit must survive + // via makeup_diff, exactly like a concurrent write in the Go server. + let mut late = Needle { + id: NeedleId(99), + cookie: Cookie(99), + data: b"late-write".to_vec(), + data_size: 10, + ..Needle::default() + }; + v.write_needle(&mut late, true, false).unwrap(); + + v.commit_compact().unwrap(); + + let mut got = Needle { + id: NeedleId(99), + cookie: Cookie(99), + ..Needle::default() + }; + v.read_needle(&mut got).unwrap(); + assert_eq!(got.data, b"late-write"); + } + /// Vacuum compaction must tolerate an .idx entry whose offset points past /// the end of the .dat file (the failure mode in issue #8928). The bad /// entry is silently dropped from the resulting .cpx; healthy needles diff --git a/weed/storage/volume_vacuum.go b/weed/storage/volume_vacuum.go index a1e2835ed..89b56b3b2 100644 --- a/weed/storage/volume_vacuum.go +++ b/weed/storage/volume_vacuum.go @@ -38,6 +38,12 @@ func isSkippableNeedleReadError(err error) bool { errors.Is(err, needle.ErrorCorrupted) } +// exceedsExpectedCompactedSize reports whether the compacted .dat is short of +// the live bytes the pre-compaction index snapshot expected. +func exceedsExpectedCompactedSize(expectedLiveBytes uint64, dstDatSize int64) bool { + return expectedLiveBytes > uint64(dstDatSize) +} + type ProgressFunc func(processed int64) bool func (v *Volume) garbageLevel() float64 { @@ -674,8 +680,9 @@ func (v *Volume) copyDataBasedOnIndexFile(opts *CompactOptions) (err error) { writeThrottler := util.NewWriteThrottler(opts.MaxBytesPerSecond) var ( - skippedNeedles int - skippedDataBytes uint64 + skippedNeedles int + skippedDataBytes uint64 + expectedLiveBytes uint64 ) err = oldNm.AscendingVisit(func(value needle_map.NeedleValue) error { @@ -715,6 +722,8 @@ func (v *Volume) copyDataBasedOnIndexFile(opts *CompactOptions) (err error) { return nil } + expectedLiveBytes += uint64(size) + if err = newNm.Set(n.Id, ToOffset(newOffset), n.Size); err != nil { return fmt.Errorf("cannot put needle: %s", err) } @@ -735,28 +744,17 @@ func (v *Volume) copyDataBasedOnIndexFile(opts *CompactOptions) (err error) { glog.Warningf("vacuum volume %d: dropped %d unreadable index entries (%d data bytes) during compaction", v.Id, skippedNeedles, skippedDataBytes) } - if v.Ttl.String() == "" && v.nm != nil { + if v.Ttl.String() == "" { dstDatSize, _, err := dstDatBackend.GetStat() if err != nil { return err } - if v.nm.ContentSize() > v.nm.DeletedSize() { - expectedContentSize := v.nm.ContentSize() - v.nm.DeletedSize() - // Skipped needles still contribute to the source-side ContentSize but - // were not written to the destination, so subtract them before the - // safety check to avoid a false positive. - if skippedDataBytes >= expectedContentSize { - expectedContentSize = 0 - } else { - expectedContentSize -= skippedDataBytes - } - if expectedContentSize > uint64(dstDatSize) { - return fmt.Errorf("volume %s unexpected new data size: %d does not match size of content minus deleted: %d", - v.Id.String(), dstDatSize, expectedContentSize) - } - } else if v.nm.DeletedSize() > v.nm.ContentSize() { - glog.Warningf("volume %s content size: %d less deleted size: %d, new size: %d", - v.Id.String(), v.nm.ContentSize(), v.nm.DeletedSize(), dstDatSize) + // expectedLiveBytes is tallied from oldNm (the frozen snapshot this + // loop copied from), not the live v.nm; unreadable needles already + // return before the tally, so no further skipped-byte adjustment. + if exceedsExpectedCompactedSize(expectedLiveBytes, dstDatSize) { + return fmt.Errorf("volume %s unexpected new data size: %d does not match expected live content size %d from the pre-compaction snapshot", + v.Id.String(), dstDatSize, expectedLiveBytes) } } err = newNm.SaveToIdx(opts.destIdxPath) diff --git a/weed/storage/volume_vacuum_test.go b/weed/storage/volume_vacuum_test.go index d56edd253..03ffcaaad 100644 --- a/weed/storage/volume_vacuum_test.go +++ b/weed/storage/volume_vacuum_test.go @@ -367,6 +367,72 @@ func TestCompactByIndex_DropsDanglingNeedle(t *testing.T) { v.Close() } +// TestCompactByIndex_ConcurrentWriteDoesNotFailIntegrityCheck reproduces the +// vacuum-vs-live-traffic race: a needle written to the volume after +// CompactByIndex has already loaded its point-in-time index snapshot must not +// trip the post-copy integrity check. CommitCompact's makeupDiff is what +// reconciles a write landing mid-copy (see TestCommitCompactDeletionTailKeepsWritable); +// the copy-phase check must not treat that expected case as corruption. +func TestCompactByIndex_ConcurrentWriteDoesNotFailIntegrityCheck(t *testing.T) { + dir := t.TempDir() + + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + defer v.Close() + + for i := 1; i <= 8; i++ { + if _, _, _, err := v.writeNeedle2(newRandomNeedle(uint64(i)), true, false, false); err != nil { + t.Fatalf("write needle %d: %v", i, err) + } + } + + wroteConcurrently := false + opts := &CompactOptions{ + ProgressCallback: func(processed int64) bool { + if !wroteConcurrently { + wroteConcurrently = true + // Simulate a client write landing on the live volume while + // CompactByIndex is still copying the pre-write snapshot. + if _, _, _, err := v.writeNeedle2(newRandomNeedle(uint64(100)), true, false, false); err != nil { + t.Fatalf("concurrent write: %v", err) + } + } + return true + }, + } + + if err := v.CompactByIndex(opts); err != nil { + t.Fatalf("CompactByIndex should tolerate a write that lands mid-copy, got: %v", err) + } +} + +// TestExceedsExpectedCompactedSize guards the copy-phase integrity check +// against regressing into double-subtracting skipped bytes: expectedLiveBytes +// already excludes needles dropped as unreadable (they return before being +// added to the tally), so the check must compare it directly against the +// compacted .dat size, with no further adjustment for skipped bytes. +func TestExceedsExpectedCompactedSize(t *testing.T) { + cases := []struct { + name string + expectedLiveBytes uint64 + dstDatSize int64 + wantExceeds bool + }{ + {"destination matches expected size exactly", 100, 100, false}, + {"destination larger than expected is fine", 100, 150, false}, + {"destination short of expected signals data loss", 100, 90, true}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := exceedsExpectedCompactedSize(c.expectedLiveBytes, c.dstDatSize); got != c.wantExceeds { + t.Fatalf("exceedsExpectedCompactedSize(%d, %d) = %v, want %v", c.expectedLiveBytes, c.dstDatSize, got, c.wantExceeds) + } + }) + } +} + func doSomeWritesDeletes(i int, v *Volume, t *testing.T, infos []*needleInfo) { n := newRandomNeedle(uint64(i)) _, size, _, err := v.writeNeedle2(n, true, false, false)