mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-20 13:30:46 +02:00
fix(vacuum): stop comparing compact size against the live needle map (#11263)
* fix(vacuum): stop comparing compact size against the live needle map CompactByIndex's post-copy integrity check compared bytes written to the .cpd against v.nm.ContentSize()-DeletedSize(), the live map that keeps mutating for as long as the volume stays writable during the copy. Any write landing after the point-in-time index snapshot was loaded made the live map's tally exceed what got copied, aborting compaction with "unexpected new data size" — even though CommitCompact's makeupDiff exists specifically to reconcile writes that land mid-copy. On a busy volume this can fail every vacuum cycle. Tally the expected live size from oldNm, the same frozen snapshot the copy loop reads from, instead of the live map. This keeps the check's original protection (destination smaller than what should have been copied signals real data loss) while removing the false positive from ordinary concurrent traffic. * fix(vacuum): stop double-subtracting skipped bytes from the size check Unreadable needles return before reaching the expectedLiveBytes tally, so it already excludes them. Subtracting skippedDataBytes again on top loosened the integrity check's margin by that same amount, letting a .cpd short of the true expected size slip past undetected — the exact failure mode the check exists to catch. Flagged independently by three automated PR reviewers (Devin, Greptile, CodeRabbit). Extract the comparison into exceedsExpectedCompactedSize and drop the subtraction entirely; add TestExceedsExpectedCompactedSize to pin the threshold to expectedLiveBytes alone. * fix(vacuum): trim verbose integrity-check comment Reduce the 8-line block comment to a concise 3-line rationale. No behavior change. * fix(vacuum): mirror compact integrity check in Rust volume server Mirror the Go fix in the Rust volume server's do_compact_by_index: tally expected_live_bytes from the frozen index snapshot (not the live needle map) and compare the compacted .dat against it after the copy. Unreadable needles already return before the tally, so no skipped-byte adjustment is needed. Adds exceeds_expected_compacted_size and two regression tests. * fix(vacuum): exercise makeup_diff in Rust concurrent-write test Address CodeRabbit review: write a needle after compaction (before commit), then call commit_compact() and assert the late write survives via makeup_diff. This actually exercises the concurrent-write path rather than just confirming the integrity check passes. --------- Co-authored-by: chrislusf <chris.lu@gmail.com>
This commit is contained in:
Da.Sanchez
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user