From 5a54beac809d7e4c2855c0643b8e336511c4bff2 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 21 Jul 2026 08:59:14 -0700 Subject: [PATCH] EC decode: read shards with the encode-time block layout (#10385) * erasure_coding: WriteDatFile takes the encode-time dat size for the shard block layout * volume server: derive EC decode layout from the encode-time dat size, not the live extent * erasure_coding: test decode after tail deletions shrink the live extent below a large-block row * seaweed-volume: write_dat_file_from_shards takes the encode-time dat size for the shard block layout * seaweed-volume: derive EC decode layout from the encode-time dat size, not the live extent * seaweed-volume: test decode after tail deletions shrink the live extent below a large-block row * erasure_coding: reject decoding with no data shards * worker: record the encode-time dat size in the .vif * erasure_coding: fall back to the shard-derived layout only when the encode-time dat size is missing * erasure_coding: reject an ambiguous shard-derived block layout * seaweed-volume: fall back to the shard-derived layout only when the encode-time dat size is missing * seaweed-volume: reject an ambiguous shard-derived block layout --- seaweed-volume/src/server/grpc_server.rs | 7 + .../src/storage/erasure_coding/ec_decoder.rs | 340 +++++++++++++++++- .../src/storage/erasure_coding/ec_encoder.rs | 9 +- weed/command/fix.go | 2 +- weed/server/volume_grpc_erasure_coding.go | 6 +- weed/storage/erasure_coding/ec_decoder.go | 54 ++- .../storage/erasure_coding/ec_decoder_test.go | 2 +- .../erasure_coding/ec_roundtrip_test.go | 119 +++++- weed/worker/tasks/erasure_coding/ec_task.go | 8 + 9 files changed, 516 insertions(+), 31 deletions(-) diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index e0993de21..c695ddcb1 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -3152,6 +3152,7 @@ impl VolumeServer for VolumeGrpcService { let dat_dir = ec_vol.dir.clone(); let ecx_dir = ec_vol.ecx_actual_dir().to_string(); let collection = ec_vol.collection.clone(); + let vif_dat_file_size = ec_vol.dat_file_size; // shard_dirs[i] is guaranteed Some for i in 0..data_shards by // the check above; collect concrete dirs for the decoder. let per_shard_dirs: Vec = shard_dirs[..data_shards] @@ -3171,12 +3172,18 @@ impl VolumeServer for VolumeGrpcService { ) .map_err(|e| Status::internal(format!("FindDatFileSize: {}", e)))?; + // The shard block layout was fixed by the .dat size at encode time + // (recorded in .vif); deletions can shrink the live extent below a + // large-block row boundary, so the layout must not be derived from + // dat_file_size. The decoder infers the layout from the shard size + // when .vif does not record it. // Write .dat file using block-interleaved reading from shards. crate::storage::erasure_coding::ec_decoder::write_dat_file_from_shards_with_dirs( &dat_dir, &collection, vid, dat_file_size, + vif_dat_file_size, data_shards, &per_shard_dirs, ) diff --git a/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs b/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs index 6bcb0ada6..428184538 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs @@ -78,6 +78,7 @@ pub fn write_dat_file_from_shards( collection: &str, volume_id: VolumeId, dat_file_size: i64, + encoded_dat_file_size: i64, data_shards: usize, ) -> io::Result<()> { let dirs: Vec = (0..data_shards).map(|_| dir.to_string()).collect(); @@ -86,6 +87,7 @@ pub fn write_dat_file_from_shards( collection, volume_id, dat_file_size, + encoded_dat_file_size, data_shards, &dirs, ) @@ -99,18 +101,59 @@ pub fn write_dat_file_from_shards( /// one dir" case both can be the same value. /// /// Mirrors Go's `WriteDatFile(baseFileName, datFileSize, -/// shardFileNames)` shape — Go passes per-shard paths so a -/// reconciled volume with shards split across disks of the same -/// volume server can still be decoded back to a regular .dat +/// encodedDatFileSize, shardFileNames)` shape — Go passes per-shard +/// paths so a reconciled volume with shards split across disks of the +/// same volume server can still be decoded back to a regular .dat /// (seaweedfs/seaweedfs#9252). +/// +/// `dat_file_size` is the number of bytes to write, i.e. the live data +/// extent from [`find_dat_file_size`]. `encoded_dat_file_size` is the +/// .dat size at encode time, which fixed the shard block layout: +/// deletions can move the live extent below the large-block row +/// boundary, and deriving the layout from the shrunk extent would read +/// the shards in the wrong block order. Pass zero when the .vif does +/// not record the encode-time size to infer the layout from the shard +/// size. pub fn write_dat_file_from_shards_with_dirs( dat_dir: &str, collection: &str, volume_id: VolumeId, dat_file_size: i64, + encoded_dat_file_size: i64, data_shards: usize, shard_dirs: &[String], ) -> io::Result<()> { + write_dat_file( + dat_dir, + collection, + volume_id, + dat_file_size, + encoded_dat_file_size, + data_shards, + shard_dirs, + ERASURE_CODING_LARGE_BLOCK_SIZE, + ERASURE_CODING_SMALL_BLOCK_SIZE, + ) +} + +#[allow(clippy::too_many_arguments)] +fn write_dat_file( + dat_dir: &str, + collection: &str, + volume_id: VolumeId, + dat_file_size: i64, + encoded_dat_file_size: i64, + data_shards: usize, + shard_dirs: &[String], + large_block_size: usize, + small_block_size: usize, +) -> io::Result<()> { + if data_shards == 0 { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "no data shards", + )); + } if shard_dirs.len() < data_shards { return Err(io::Error::new( io::ErrorKind::InvalidInput, @@ -138,36 +181,82 @@ pub fn write_dat_file_from_shards_with_dirs( shard.open()?; } + let mut encoded_dat_file_size = encoded_dat_file_size; + if encoded_dat_file_size <= 0 { + // .vif without the encode-time size: infer the padded layout from + // the physical shard size, which reads the shards in the same + // block order. + let shard_size = std::fs::metadata(shards[0].file_name())?.len() as i64; + // A shard size that is an exact multiple of the large block size + // is ambiguous: N large rows, or N-1 large rows plus a full + // small-block region. The two layouts only agree below the last + // large row. + let large = large_block_size as i64; + if shard_size % large == 0 + && dat_file_size > (shard_size / large - 1) * large * data_shards as i64 + { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "shard size {} does not identify the block layout; re-encode to record the dat size in .vif", + shard_size + ), + )); + } + encoded_dat_file_size = data_shards as i64 * shard_size; + } + if dat_file_size > encoded_dat_file_size { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!( + "dat file size {} exceeds encoded dat file size {}", + dat_file_size, encoded_dat_file_size + ), + )); + } + let mut dat_file = File::create(&tmp_path)?; let mut remaining = dat_file_size; - let large_block_size = ERASURE_CODING_LARGE_BLOCK_SIZE; - let small_block_size = ERASURE_CODING_SMALL_BLOCK_SIZE; + let mut encoded_remaining = encoded_dat_file_size; let large_row_size = (large_block_size * data_shards) as i64; let mut shard_offset: u64 = 0; // Read large blocks - while remaining >= large_row_size { + while encoded_remaining >= large_row_size && remaining > 0 { for i in 0..data_shards { - let mut buf = vec![0u8; large_block_size]; - shards[i].read_at(&mut buf, shard_offset)?; let to_write = large_block_size.min(remaining as usize); - dat_file.write_all(&buf[..to_write])?; + let mut buf = vec![0u8; to_write]; + let n = shards[i].read_at(&mut buf, shard_offset)?; + if n != to_write { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + format!("short read of large block on shard {}", i), + )); + } + dat_file.write_all(&buf)?; remaining -= to_write as i64; if remaining <= 0 { break; } } + encoded_remaining -= large_row_size; shard_offset += large_block_size as u64; } // Read small blocks while remaining > 0 { for i in 0..data_shards { - let mut buf = vec![0u8; small_block_size]; - shards[i].read_at(&mut buf, shard_offset)?; let to_write = small_block_size.min(remaining as usize); - dat_file.write_all(&buf[..to_write])?; + let mut buf = vec![0u8; to_write]; + let n = shards[i].read_at(&mut buf, shard_offset)?; + if n != to_write { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + format!("short read of small block on shard {}", i), + )); + } + dat_file.write_all(&buf)?; remaining -= to_write as i64; if remaining <= 0 { break; @@ -330,8 +419,15 @@ mod tests { std::fs::remove_file(format!("{}/1.idx", dir)).unwrap(); // Reconstruct from EC shards - write_dat_file_from_shards(dir, "", VolumeId(1), original_dat_size as i64, data_shards) - .unwrap(); + write_dat_file_from_shards( + dir, + "", + VolumeId(1), + original_dat_size as i64, + original_dat_size as i64, + data_shards, + ) + .unwrap(); write_idx_file_from_ec_index(dir, "", VolumeId(1)).unwrap(); // Atomic publish must rename the temp files away, never leaving them behind. @@ -376,9 +472,223 @@ mod tests { let dir = tmp.path().to_str().unwrap(); // No shard files exist, so de-striping must fail and publish nothing: // neither the final .dat nor a partial .dat.tmp may remain. - let res = write_dat_file_from_shards(dir, "", VolumeId(7), 100, 10); + let res = write_dat_file_from_shards(dir, "", VolumeId(7), 100, 100, 10); assert!(res.is_err()); assert!(!std::path::Path::new(&format!("{}/7.dat", dir)).exists()); assert!(!std::path::Path::new(&format!("{}/7.dat.tmp", dir)).exists()); } + + + // Decoding when .vif does not record the encode-time size: the layout is + // inferred from the shard size, except when that is an exact large-block + // multiple and the live extent reaches the ambiguous region. + #[test] + fn test_write_dat_file_fallback_layout() { + use crate::storage::erasure_coding::ec_bitrot::{ + ShardChecksumBuilder, DEFAULT_BITROT_BLOCK_SIZE, + }; + use reed_solomon_erasure::galois_8::ReedSolomon; + + const LARGE: usize = 10000; + const SMALL: usize = 100; + let data_shards = 10usize; + let parity_shards = 4usize; + let large_row_size = (LARGE * data_shards) as i64; + + let tmp = TempDir::new().unwrap(); + + let encode = |name: &str, dat_size: i64| -> (String, Vec, Vec) { + let dir = tmp.path().join(name); + std::fs::create_dir(&dir).unwrap(); + let dir = dir.to_str().unwrap().to_string(); + let original: Vec = (0..dat_size as usize) + .map(|i| ((i as u64).wrapping_mul(2654435761) >> 8) as u8) + .collect(); + std::fs::write(format!("{}/1.dat", dir), &original).unwrap(); + let dat_file = File::open(format!("{}/1.dat", dir)).unwrap(); + let rs = ReedSolomon::new(data_shards, parity_shards).unwrap(); + let total = data_shards + parity_shards; + let mut shards: Vec = (0..total as u8) + .map(|i| EcVolumeShard::new(&dir, "", VolumeId(1), i)) + .collect(); + for shard in &mut shards { + shard.create().unwrap(); + } + let mut builders: Vec = (0..total) + .map(|_| ShardChecksumBuilder::new(DEFAULT_BITROT_BLOCK_SIZE as i64)) + .collect(); + ec_encoder::encode_dat_file( + &dat_file, + dat_size, + &rs, + &mut shards, + &mut builders, + data_shards, + parity_shards, + LARGE, + SMALL, + ) + .unwrap(); + for shard in &mut shards { + shard.close(); + } + let shard_dirs: Vec = (0..data_shards).map(|_| dir.clone()).collect(); + (dir, shard_dirs, original) + }; + + let decode_to = |dir: &str, + sub: &str, + live: i64, + encoded: i64, + shard_dirs: &[String]| + -> io::Result> { + let out = format!("{}/{}", dir, sub); + std::fs::create_dir_all(&out).unwrap(); + write_dat_file(&out, "", VolumeId(1), live, encoded, 10, shard_dirs, LARGE, SMALL)?; + Ok(std::fs::read(format!("{}/1.dat", out)).unwrap()) + }; + + // a small-block tail that is not a large-block multiple is unambiguous + let (dir, shard_dirs, original) = encode("plain", large_row_size + 2530); + let live = large_row_size / 2; + let decoded = decode_to(&dir, "out", live, 0, &shard_dirs).unwrap(); + assert_eq!(&original[..live as usize], &decoded[..]); + + // datSize just under one large row: the full small-block region makes + // each shard exactly one large block, indistinguishable from one large row + let (dir, shard_dirs, _) = encode("ambig1", large_row_size - 1); + let err = decode_to(&dir, "out", large_row_size / 2, 0, &shard_dirs).unwrap_err(); + assert!(err.to_string().contains("does not identify the block layout")); + + // two-row equivalent: decoding within the agreed prefix still works + let (dir, shard_dirs, original) = encode("ambig2", 2 * large_row_size - 1); + let decoded = decode_to(&dir, "outa", large_row_size, 0, &shard_dirs).unwrap(); + assert_eq!(&original[..large_row_size as usize], &decoded[..]); + let err = decode_to(&dir, "outb", large_row_size + 1, 0, &shard_dirs).unwrap_err(); + assert!(err.to_string().contains("does not identify the block layout")); + } + + // Decoding after deletions moved the live extent below the large-block row + // boundary: the shard block layout is fixed by the encode-time .dat size, + // so de-striping must not derive it from the shrunk live extent. + #[test] + fn test_write_dat_file_after_tail_deletion() { + use crate::storage::erasure_coding::ec_bitrot::{ + ShardChecksumBuilder, DEFAULT_BITROT_BLOCK_SIZE, + }; + use reed_solomon_erasure::galois_8::ReedSolomon; + + const LARGE: usize = 10000; + const SMALL: usize = 100; + let data_shards = 10usize; + let parity_shards = 4usize; + let large_row_size = (LARGE * data_shards) as i64; + let small_row_size = (SMALL * data_shards) as i64; + + // one full large-block row plus a small-block tail + let dat_size = large_row_size + 2 * small_row_size + 530; + + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + + let original: Vec = (0..dat_size as usize) + .map(|i| ((i as u64).wrapping_mul(2654435761) >> 8) as u8) + .collect(); + std::fs::write(format!("{}/1.dat", dir), &original).unwrap(); + + let dat_file = File::open(format!("{}/1.dat", dir)).unwrap(); + let rs = ReedSolomon::new(data_shards, parity_shards).unwrap(); + let total = data_shards + parity_shards; + let mut shards: Vec = (0..total as u8) + .map(|i| EcVolumeShard::new(dir, "", VolumeId(1), i)) + .collect(); + for shard in &mut shards { + shard.create().unwrap(); + } + let mut builders: Vec = (0..total) + .map(|_| ShardChecksumBuilder::new(DEFAULT_BITROT_BLOCK_SIZE as i64)) + .collect(); + ec_encoder::encode_dat_file( + &dat_file, + dat_size, + &rs, + &mut shards, + &mut builders, + data_shards, + parity_shards, + LARGE, + SMALL, + ) + .unwrap(); + for shard in &mut shards { + shard.close(); + } + + let shard_size = std::fs::metadata(format!("{}/1.ec00", dir)).unwrap().len() as i64; + let padded_size = data_shards as i64 * shard_size; + let shard_dirs: Vec = (0..data_shards).map(|_| dir.to_string()).collect(); + + // decode into a separate dir so the output does not collide with the source .dat + let out_dir = tmp.path().join("out"); + std::fs::create_dir(&out_dir).unwrap(); + let out = out_dir.to_str().unwrap(); + let decode = |live_size: i64, encoded_size: i64| -> Vec { + write_dat_file( + out, + "", + VolumeId(1), + live_size, + encoded_size, + data_shards, + &shard_dirs, + LARGE, + SMALL, + ) + .unwrap(); + let path = format!("{}/1.dat", out); + let decoded = std::fs::read(&path).unwrap(); + std::fs::remove_file(&path).unwrap(); + decoded + }; + + for live_size in [ + LARGE as i64 - 1, // within the first large block + LARGE as i64 + 42, // partial second large block + large_row_size / 2, // mid large row + large_row_size, // exactly the large row + large_row_size + 5 * SMALL as i64, // into the small-block tail + dat_size, // nothing deleted + ] { + assert_eq!( + &original[..live_size as usize], + &decode(live_size, dat_size)[..], + "live size {} with encode-time layout", + live_size + ); + assert_eq!( + &original[..live_size as usize], + &decode(live_size, padded_size)[..], + "live size {} with padded layout", + live_size + ); + } + + // deriving the layout from the shrunk live extent reorders the data + let control = decode(large_row_size / 2, large_row_size / 2); + assert_ne!(&original[..(large_row_size / 2) as usize], &control[..]); + + // the live extent can never exceed the encode-time size + assert!(write_dat_file( + out, + "", + VolumeId(1), + dat_size + 1, + dat_size, + data_shards, + &shard_dirs, + LARGE, + SMALL, + ) + .is_err()); + } } diff --git a/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs b/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs index c66e63c07..15b944483 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs @@ -75,6 +75,8 @@ pub fn write_ec_files( &mut builders, data_shards, parity_shards, + ERASURE_CODING_LARGE_BLOCK_SIZE, + ERASURE_CODING_SMALL_BLOCK_SIZE, )?; // Close all shards @@ -582,7 +584,8 @@ fn read_from_data_shards( /// Uses a two-phase approach matching Go's ec_encoder.go: /// 1. Process as many large blocks (1GB) as possible /// 2. Process remaining data with small blocks (1MB) -fn encode_dat_file( +#[allow(clippy::too_many_arguments)] +pub(crate) fn encode_dat_file( dat_file: &File, dat_size: i64, rs: &ReedSolomon, @@ -590,12 +593,13 @@ fn encode_dat_file( builders: &mut [ShardChecksumBuilder], data_shards: usize, parity_shards: usize, + large_block_size: usize, + small_block_size: usize, ) -> io::Result<()> { let mut remaining = dat_size; let mut offset: u64 = 0; // Phase 1: Process large blocks (1GB each) while enough data remains - let large_block_size = ERASURE_CODING_LARGE_BLOCK_SIZE; let large_row_size = large_block_size * data_shards; while remaining >= large_row_size as i64 { @@ -614,7 +618,6 @@ fn encode_dat_file( } // Phase 2: Process remaining data with small blocks (1MB each) - let small_block_size = ERASURE_CODING_SMALL_BLOCK_SIZE; let small_row_size = small_block_size * data_shards; while remaining > 0 { diff --git a/weed/command/fix.go b/weed/command/fix.go index e92f67a47..6c24dd320 100644 --- a/weed/command/fix.go +++ b/weed/command/fix.go @@ -410,7 +410,7 @@ func doFixEcxFromShards(basePath, baseFileName, collection string, volumeId int6 // De-stripe the data shards into a temporary .dat next to the shards. tmpBase := base + ".ecxrecover" tmpDat := tmpBase + ".dat" - if err := erasure_coding.WriteDatFile(tmpBase, reconstructSize, shardFileNames); err != nil { + if err := erasure_coding.WriteDatFile(tmpBase, reconstructSize, reconstructSize, shardFileNames); err != nil { os.Remove(tmpDat) fail(fmt.Errorf("volume %d: reconstruct .dat from data shards: %w", volumeId, err)) return diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index 443632287..25555a386 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -965,8 +965,12 @@ func (vs *VolumeServer) VolumeEcShardsToVolume(ctx context.Context, req *volume_ return nil, fmt.Errorf("FindDatFileSize %s: %v", dataBaseFileName, err) } + // The shard block layout was fixed by the .dat size at encode time (recorded + // in .vif); deletions can shrink the live extent below a large-block row + // boundary, so the layout must not be derived from datFileSize. WriteDatFile + // infers the layout from the shard size when .vif does not record it. // write .dat file from .ec00 ~ .ec09 files - if err := erasure_coding.WriteDatFile(dataBaseFileName, datFileSize, shardFileNames); err != nil { + if err := erasure_coding.WriteDatFile(dataBaseFileName, datFileSize, v.DatFileSize(), shardFileNames); err != nil { return nil, fmt.Errorf("WriteDatFile %s: %v", dataBaseFileName, err) } diff --git a/weed/storage/erasure_coding/ec_decoder.go b/weed/storage/erasure_coding/ec_decoder.go index 1fd890e8e..6d7c9d9b6 100644 --- a/weed/storage/erasure_coding/ec_decoder.go +++ b/weed/storage/erasure_coding/ec_decoder.go @@ -204,8 +204,22 @@ func iterateEcjFile(baseFileName string, processNeedleFn func(key types.NeedleId } -// WriteDatFile generates .dat from EC shard files (e.g., .ec00 ~ .ec09 for 10+4) -func WriteDatFile(baseFileName string, datFileSize int64, shardFileNames []string) error { +// WriteDatFile generates .dat from EC shard files (e.g., .ec00 ~ .ec09 for 10+4). +// datFileSize is the number of bytes to write, i.e. the live data extent from +// FindDatFileSize. encodedDatFileSize is the .dat size at encode time, which +// fixed the shard block layout: deletions can move the live extent below the +// large-block row boundary, and deriving the layout from the shrunk extent +// would read the shards in the wrong block order. Pass zero when the .vif does +// not record the encode-time size to infer the layout from the shard size. +func WriteDatFile(baseFileName string, datFileSize int64, encodedDatFileSize int64, shardFileNames []string) error { + return writeDatFile(baseFileName, datFileSize, encodedDatFileSize, shardFileNames, ErasureCodingLargeBlockSize, ErasureCodingSmallBlockSize) +} + +func writeDatFile(baseFileName string, datFileSize int64, encodedDatFileSize int64, shardFileNames []string, largeBlockSize int64, smallBlockSize int64) error { + + if len(shardFileNames) == 0 { + return fmt.Errorf("no data shard files") + } // Write to a temp file and atomically rename into place, so a crash mid-write // never leaves a partial .dat at the final name beside the source shards. @@ -241,19 +255,41 @@ func WriteDatFile(baseFileName string, datFileSize int64, shardFileNames []strin } } - for datFileSize >= int64(dataShards)*ErasureCodingLargeBlockSize { - for shardId := 0; shardId < dataShards; shardId++ { - w, err := io.CopyN(datFile, inputFiles[shardId], ErasureCodingLargeBlockSize) - if w != ErasureCodingLargeBlockSize { + if encodedDatFileSize <= 0 { + // .vif without the encode-time size: infer the padded layout from the + // physical shard size, which reads the shards in the same block order. + shardFileInfo, statErr := inputFiles[0].Stat() + if statErr != nil { + return fmt.Errorf("stat %s: %v", shardFileNames[0], statErr) + } + shardSize := shardFileInfo.Size() + // A shard size that is an exact multiple of the large block size is + // ambiguous: N large rows, or N-1 large rows plus a full small-block + // region. The two layouts only agree below the last large row. + if shardSize%largeBlockSize == 0 && datFileSize > (shardSize/largeBlockSize-1)*largeBlockSize*int64(dataShards) { + return fmt.Errorf("shard size %d of %s does not identify the block layout; re-encode to record the dat size in .vif", shardSize, baseFileName) + } + encodedDatFileSize = int64(dataShards) * shardSize + } + if datFileSize > encodedDatFileSize { + return fmt.Errorf("dat file size %d exceeds encoded dat file size %d", datFileSize, encodedDatFileSize) + } + + for encodedDatFileSize >= int64(dataShards)*largeBlockSize && datFileSize > 0 { + for shardId := 0; shardId < dataShards && datFileSize > 0; shardId++ { + toRead := min(datFileSize, largeBlockSize) + w, err := io.CopyN(datFile, inputFiles[shardId], toRead) + if w != toRead { return fmt.Errorf("copy %s large block on shardId %d: %v", baseFileName, shardId, err) } - datFileSize -= ErasureCodingLargeBlockSize + datFileSize -= toRead } + encodedDatFileSize -= int64(dataShards) * largeBlockSize } for datFileSize > 0 { - for shardId := 0; shardId < dataShards; shardId++ { - toRead := min(datFileSize, ErasureCodingSmallBlockSize) + for shardId := 0; shardId < dataShards && datFileSize > 0; shardId++ { + toRead := min(datFileSize, smallBlockSize) w, err := io.CopyN(datFile, inputFiles[shardId], toRead) if w != toRead { return fmt.Errorf("copy %s small block %d: %v", baseFileName, shardId, err) diff --git a/weed/storage/erasure_coding/ec_decoder_test.go b/weed/storage/erasure_coding/ec_decoder_test.go index 6793d890d..bbdbb353d 100644 --- a/weed/storage/erasure_coding/ec_decoder_test.go +++ b/weed/storage/erasure_coding/ec_decoder_test.go @@ -217,7 +217,7 @@ func TestDecodeAtomicPublish(t *testing.T) { // final .dat nor a partial .dat.tmp behind. datBase := filepath.Join(dir, "bar_2") missingShards := []string{filepath.Join(dir, "does_not_exist.ec00")} - if err := erasure_coding.WriteDatFile(datBase, 100, missingShards); err == nil { + if err := erasure_coding.WriteDatFile(datBase, 100, 100, missingShards); err == nil { t.Fatalf("expected WriteDatFile to fail on missing shard") } if _, err := os.Stat(datBase + ".dat"); !os.IsNotExist(err) { diff --git a/weed/storage/erasure_coding/ec_roundtrip_test.go b/weed/storage/erasure_coding/ec_roundtrip_test.go index 46540a182..8c07a338d 100644 --- a/weed/storage/erasure_coding/ec_roundtrip_test.go +++ b/weed/storage/erasure_coding/ec_roundtrip_test.go @@ -326,7 +326,7 @@ func testDecodeDat(t *testing.T, datSize int64) { shardFileNames[i] = fmt.Sprintf("%s%s", baseFileName, ctx.ToExt(i)) } - err = WriteDatFile(decodedBase, datSize, shardFileNames) + err = WriteDatFile(decodedBase, datSize, datSize, shardFileNames) require.NoError(t, err, "WriteDatFile") // The atomic publish must rename the temp file away, never leaving it behind. @@ -390,3 +390,120 @@ func assembleFromIntervals(ecFiles []*os.File, intervals []Interval, large, smal } return data, nil } + +// TestWriteDatFileAfterTailDeletion decodes after deletions moved the live +// extent below the large-block row boundary. The shard block layout is fixed +// by the encode-time .dat size, so de-striping must not derive it from the +// shrunk live extent. +func TestWriteDatFileAfterTailDeletion(t *testing.T) { + const ( + large = int64(largeBlockSize) // 10000 + small = int64(smallBlockSize) // 100 + ) + largeRowSize := large * DataShardsCount + smallRowSize := small * DataShardsCount + + // one full large-block row plus a small-block tail + datSize := largeRowSize + 2*smallRowSize + 530 + + dir := t.TempDir() + baseFileName := fmt.Sprintf("%s/tail_del", dir) + + originalData := make([]byte, datSize) + _, err := rand.Read(originalData) + require.NoError(t, err) + require.NoError(t, os.WriteFile(baseFileName+".dat", originalData, 0644)) + + ctx := NewDefaultECContext("", 0) + _, err = generateEcFiles(baseFileName, 50, large, small, ctx) + require.NoError(t, err, "EC encoding") + + shardFileNames := make([]string, DataShardsCount) + for i := 0; i < DataShardsCount; i++ { + shardFileNames[i] = baseFileName + ctx.ToExt(i) + } + shardFileInfo, err := os.Stat(shardFileNames[0]) + require.NoError(t, err) + paddedSize := int64(DataShardsCount) * shardFileInfo.Size() + + decodedBase := baseFileName + "_decoded" + decode := func(liveSize, encodedSize int64) []byte { + require.NoError(t, writeDatFile(decodedBase, liveSize, encodedSize, shardFileNames, large, small)) + decoded, readErr := os.ReadFile(decodedBase + ".dat") + require.NoError(t, readErr) + require.NoError(t, os.Remove(decodedBase+".dat")) + return decoded + } + + liveSizes := []int64{ + large - 1, // within the first large block + large + 42, // partial second large block + largeRowSize / 2, // mid large row + largeRowSize, // exactly the large row + largeRowSize + 5*small, // into the small-block tail + datSize, // nothing deleted + } + for _, liveSize := range liveSizes { + assert.Equal(t, originalData[:liveSize], decode(liveSize, datSize), "live size %d, encode-time layout", liveSize) + assert.Equal(t, originalData[:liveSize], decode(liveSize, paddedSize), "live size %d, padded layout", liveSize) + } + + // deriving the layout from the shrunk live extent reorders the data + assert.NotEqual(t, originalData[:largeRowSize/2], decode(largeRowSize/2, largeRowSize/2)) + + // the live extent can never exceed the encode-time size + require.Error(t, writeDatFile(decodedBase, datSize+1, datSize, shardFileNames, large, small)) +} + +// TestWriteDatFileFallbackLayout covers decoding when .vif does not record the +// encode-time size: the layout is inferred from the shard size, except when +// that is an exact large-block multiple and the live extent reaches the +// ambiguous region. +func TestWriteDatFileFallbackLayout(t *testing.T) { + const ( + large = int64(largeBlockSize) // 10000 + small = int64(smallBlockSize) // 100 + ) + largeRowSize := large * DataShardsCount + + dir := t.TempDir() + + encode := func(name string, datSize int64) (string, []string, []byte) { + baseFileName := fmt.Sprintf("%s/%s", dir, name) + originalData := make([]byte, datSize) + _, err := rand.Read(originalData) + require.NoError(t, err) + require.NoError(t, os.WriteFile(baseFileName+".dat", originalData, 0644)) + ctx := NewDefaultECContext("", 0) + _, err = generateEcFiles(baseFileName, 50, large, small, ctx) + require.NoError(t, err, "EC encoding") + shardFileNames := make([]string, DataShardsCount) + for i := 0; i < DataShardsCount; i++ { + shardFileNames[i] = baseFileName + ctx.ToExt(i) + } + return baseFileName, shardFileNames, originalData + } + + // a small-block tail that is not a large-block multiple is unambiguous + base, shards, original := encode("plain", largeRowSize+2530) + liveSize := largeRowSize / 2 + require.NoError(t, writeDatFile(base+"_d", liveSize, 0, shards, large, small)) + decoded, err := os.ReadFile(base + "_d.dat") + require.NoError(t, err) + assert.Equal(t, original[:liveSize], decoded) + + // datSize just under one large row: the full small-block region makes each + // shard exactly largeBlockSize, indistinguishable from one large row + base, shards, _ = encode("ambiguous1", largeRowSize-1) + err = writeDatFile(base+"_d", largeRowSize/2, 0, shards, large, small) + require.ErrorContains(t, err, "does not identify the block layout") + + // two-row equivalent: decoding within the agreed prefix still works + base, shards, original = encode("ambiguous2", 2*largeRowSize-1) + require.NoError(t, writeDatFile(base+"_d", largeRowSize, 0, shards, large, small)) + decoded, err = os.ReadFile(base + "_d.dat") + require.NoError(t, err) + assert.Equal(t, original[:largeRowSize], decoded) + err = writeDatFile(base+"_d2", largeRowSize+1, 0, shards, large, small) + require.ErrorContains(t, err, "does not identify the block layout") +} diff --git a/weed/worker/tasks/erasure_coding/ec_task.go b/weed/worker/tasks/erasure_coding/ec_task.go index f37cd0228..d0372da35 100644 --- a/weed/worker/tasks/erasure_coding/ec_task.go +++ b/weed/worker/tasks/erasure_coding/ec_task.go @@ -647,6 +647,14 @@ func (t *ErasureCodingTask) generateEcShardsLocally(localFiles map[string]string Version: uint32(needle.GetCurrentVersion()), EcShardConfig: ecShardConfig, } + // The decoder resolves the shard block layout from the encode-time .dat + // size; without it, decoding falls back to inferring the layout from the + // shard size, which is ambiguous when that is a large-block multiple. + if info, err := os.Stat(datFile); err == nil { + volumeInfo.DatFileSize = info.Size() + } else { + glog.Warningf("stat %s for .vif dat file size: %v", datFile, err) + } if err := volume_info.SaveVolumeInfo(vifFile, volumeInfo); err != nil { glog.Warningf("Failed to create .vif file: %v", err) } else {