fix(storage): validate sibling .dat size before deleting partial EC (#9478)

The earlier prune deleted partial EC files whenever any .dat for the
same vid existed on a sibling disk — including a zero-byte shell. A
shell is no more useful than the partial shard it would replace, and
the partial shard might still combine with shards on other servers
in a recoverable distributed-EC layout. Wiping it based on a corrupt
sibling .dat is data loss masquerading as cleanup.

Tighten the check: when the EC's .vif recorded a non-zero source size
in datFileSize, require the sibling .dat to be at least that many
bytes; otherwise fall back to "at least a superblock". The .vif value
is what the encoder wrote at the moment the source was sealed, so a
sibling .dat smaller than that is provably truncated. Carry the size
through indexDatOwners alongside the location.

The Rust port had the same gap and an additional bug behind it:
EcVolume::new wasn't reading datFileSize from .vif, so the safety
check always fell back to the superblock floor. Wire datFileSize
through. The existing shard-size calculation in
LocateEcShardNeedleInterval already uses dat_file_size when non-zero,
so populating it also matches Go's behaviour there.

Tests cover the truncated-sibling case in both ports.
This commit is contained in:
Chris Lu 2026-05-13 00:33:59 -07:00
parent 0778d4820d
commit 36768a6ca2
5 changed files with 339 additions and 38 deletions

View File

@ -120,10 +120,15 @@ impl EcVolume {
shards.push(None);
}
// Read expire_at_sec and version from .vif if present (matches Go's MaybeLoadVolumeInfo).
// Prefer the data dir; fall back to the idx dir for the
// cross-disk reconcile case (#9212 / #9244).
let (expire_at_sec, vif_version) = {
// Read expire_at_sec, version, and dat_file_size from .vif if
// present (matches Go's MaybeLoadVolumeInfo). Prefer the data
// dir; fall back to the idx dir for the cross-disk reconcile
// case (#9212 / #9244). `dat_file_size` is the source .dat
// size at encode time, used both by `locate_ec_shard_needle`
// for shard-size math and by the Store-level prune in
// `store_ec_reconcile.rs` to verify a sibling-disk .dat is
// plausibly the encoding source (#9478).
let (expire_at_sec, vif_version, vif_dat_file_size) = {
let vif_path = locate_vif_path(dir, dir_idx, collection, volume_id);
if let Ok(vif_content) = std::fs::read_to_string(&vif_path) {
if let Ok(vif_info) =
@ -134,12 +139,12 @@ impl EcVolume {
} else {
Version::current()
};
(vif_info.expire_at_sec, ver)
(vif_info.expire_at_sec, ver, vif_info.dat_file_size)
} else {
(0, Version::current())
(0, Version::current(), 0)
}
} else {
(0, Version::current())
(0, Version::current(), 0)
}
};
@ -150,7 +155,7 @@ impl EcVolume {
dir_idx: dir_idx.to_string(),
version: vif_version,
shards,
dat_file_size: 0,
dat_file_size: vif_dat_file_size,
data_shards,
parity_shards,
ecx_file: None,

View File

@ -23,8 +23,20 @@ use tracing::{info, warn};
use crate::storage::disk_location::{is_ec_shard_extension, parse_collection_volume_id_pub};
use crate::storage::erasure_coding::ec_shard::DATA_SHARDS_COUNT;
use crate::storage::store::Store;
use crate::storage::super_block::SUPER_BLOCK_SIZE;
use crate::storage::types::VolumeId;
/// Sibling-disk `.dat` candidate for `prune_incomplete_ec_with_sibling_dat`.
/// We record both the disk index and the file's size: the size is
/// consulted before deleting any EC artefacts. A zero-byte or truncated
/// `.dat` is not a credible fallback, and we'd rather leave the partial
/// EC in place than wipe it based on garbage.
#[derive(Clone, Copy, Debug)]
struct DatOwnerInfo {
location: usize,
size: u64,
}
/// Key for orphan-shard reconciliation: collection + volume id. Two
/// collections can re-use the same volume id, and we must only pair
/// shards with their own `.ecx`.
@ -160,9 +172,18 @@ impl Store {
/// this server) also fall through unchanged because the `.dat`
/// index never matches.
///
/// We don't have to push anything to a deleted-shards channel here:
/// the Rust heartbeat path in `server/heartbeat.rs` diffs the
/// current `ec_volumes` snapshot against the previous one each
/// Before deleting any EC files we also check that the sibling
/// `.dat` is plausibly the encoding source: at least
/// `SUPER_BLOCK_SIZE` bytes long, and — when the EC's `.vif`
/// recorded a non-zero source size in `dat_file_size` — at least
/// that many bytes. A zero-byte shell or a truncated `.dat` does
/// not justify wiping the partial EC, because that EC shard may
/// still combine usefully with shards on other servers in a
/// recoverable distributed-EC layout.
///
/// We don't have to push anything to a deleted-shards channel
/// here: the Rust heartbeat path in `server/heartbeat.rs` diffs
/// the current `ec_volumes` snapshot against the previous one each
/// tick, so the first heartbeat after the prune naturally emits a
/// delete delta for whatever the per-disk pass had already
/// reported.
@ -196,20 +217,42 @@ impl Store {
collection: ev.collection.clone(),
vid: *vid,
};
let Some(owner_idx) = dat_owners.get(&key) else {
let Some(owner) = dat_owners.get(&key) else {
continue;
};
if *owner_idx == loc_idx {
if owner.location == loc_idx {
// Same-disk .dat is the job of
// DiskLocation::validate_ec_volume during the
// per-disk pass; don't second-guess it here.
continue;
}
// Decide whether the sibling .dat is credible. Prefer
// the size baked into .vif at encode time; fall back
// to "at least a superblock" for old EC volumes whose
// .vif predates the field.
let required = if ev.dat_file_size > 0 {
ev.dat_file_size as u64
} else {
SUPER_BLOCK_SIZE as u64
};
if owner.size < required {
warn!(
volume_id = vid.0,
collection = %ev.collection,
directory = %loc.directory,
shard_count,
sibling_dir = %self.locations[owner.location].directory,
sibling_dat_size = owner.size,
required,
"sibling .dat is smaller than the EC source size; leaving partial EC in place so distributed reconstruction is still possible (issue 9478)",
);
continue;
}
victims.push(Victim {
loc_idx,
collection: ev.collection.clone(),
vid: *vid,
dat_dir: self.locations[*owner_idx].directory.clone(),
dat_dir: self.locations[owner.location].directory.clone(),
shard_count,
});
}
@ -244,18 +287,20 @@ impl Store {
}
/// Returns, for every `(collection, vid)`, the index of the first
/// disk on this store that holds a `.dat` for it. Used by
/// `prune_incomplete_ec_with_sibling_dat` so it can decide whether
/// partial EC artefacts on another disk are leftovers of an
/// interrupted encode.
/// disk on this store that holds a `.dat` for it plus the file's
/// size. Used by `prune_incomplete_ec_with_sibling_dat` so it can
/// decide whether partial EC artefacts on another disk are
/// leftovers of an interrupted encode AND whether the sibling
/// `.dat` is large enough to be a credible fallback.
///
/// Any `.dat` os::read_dir can see counts — including zero-byte
/// Any `.dat` `read_dir` can see is recorded — including zero-byte
/// shells. Their mere presence means this volume was a regular
/// volume on this server at some point, which rules out the
/// "distributed EC, no .dat anywhere" reading that would justify
/// keeping the partial shards.
fn index_dat_owners(&self) -> HashMap<EcKey, usize> {
let mut owners: HashMap<EcKey, usize> = HashMap::new();
/// "distributed EC, no .dat anywhere" reading. Whether the file is
/// actually usable is the caller's call, made by comparing this
/// size to the EC's recorded source size in `.vif`.
fn index_dat_owners(&self) -> HashMap<EcKey, DatOwnerInfo> {
let mut owners: HashMap<EcKey, DatOwnerInfo> = HashMap::new();
for (loc_idx, loc) in self.locations.iter().enumerate() {
let Ok(read) = fs::read_dir(&loc.directory) else {
continue;
@ -271,7 +316,15 @@ impl Store {
let Some((collection, vid)) = parse_collection_volume_id_pub(base) else {
continue;
};
owners.entry(EcKey { collection, vid }).or_insert(loc_idx);
let Ok(meta) = ent.metadata() else {
continue;
};
owners
.entry(EcKey { collection, vid })
.or_insert(DatOwnerInfo {
location: loc_idx,
size: meta.len(),
});
}
}
owners
@ -953,6 +1006,111 @@ mod tests {
);
}
/// Safety guard for `prune_incomplete_ec_with_sibling_dat`: when
/// the sibling `.dat` is smaller than the source size recorded in
/// the EC's `.vif`, the `.dat` is clearly truncated and is not a
/// credible fallback. The partial shard may still combine with
/// shards on other servers in a recoverable distributed-EC layout,
/// so we leave the EC files in place.
#[test]
fn test_prune_keeps_partial_ec_when_sibling_dat_is_smaller_than_vif_source_size() {
let tmp = TempDir::new().unwrap();
let dat_dir = tmp.path().join("sdd");
let ec_dir = tmp.path().join("sdf");
std::fs::create_dir_all(&dat_dir).unwrap();
std::fs::create_dir_all(&ec_dir).unwrap();
let collection = "logs";
let vid = 122u32;
// Tiny but loadable .dat on sdd: a valid 8-byte version-3
// superblock so Volume::new succeeds, padded out to a few
// hundred bytes so we have something well below the .vif-
// recorded source size below. Anything smaller would let
// Volume::new's auto-write extend the file to SUPER_BLOCK_SIZE
// and obscure the truncation we're trying to model.
let dat_path = dat_dir.join(format!("{}_{}.dat", collection, vid));
let mut dat_bytes = crate::storage::super_block::SuperBlock::default().to_bytes();
dat_bytes.resize(256, 0u8);
std::fs::write(&dat_path, &dat_bytes).unwrap();
let sibling_dat_size = dat_bytes.len() as u64;
let source_size = 10 * 1024 * 1024u64; // 10 MiB recorded in .vif
assert!(sibling_dat_size < source_size);
// Partial EC: one shard plus .ecx / .ecj / .vif. The .vif
// records the source .dat size at encode time; the safety
// check compares the sibling .dat against this value.
write_shard(ec_dir.to_str().unwrap(), collection, vid, 1);
let vif = VifVolumeInfo {
version: 3,
dat_file_size: source_size as i64,
ec_shard_config: Some(VifEcShardConfig {
data_shards: 10,
parity_shards: 4,
}),
..Default::default()
};
std::fs::write(
ec_dir.join(format!("{}_{}.vif", collection, vid)),
serde_json::to_string(&vif).unwrap(),
)
.unwrap();
std::fs::write(
ec_dir.join(format!("{}_{}.ecx", collection, vid)),
vec![0u8; 20],
)
.unwrap();
std::fs::write(
ec_dir.join(format!("{}_{}.ecj", collection, vid)),
b"",
)
.unwrap();
let mut store = Store::new(NeedleMapKind::InMemory);
store
.add_location(
dat_dir.to_str().unwrap(),
dat_dir.to_str().unwrap(),
100,
DiskType::HardDrive,
MinFreeSpace::Percent(0.0),
Vec::new(),
)
.unwrap();
store
.add_location(
ec_dir.to_str().unwrap(),
ec_dir.to_str().unwrap(),
100,
DiskType::HardDrive,
MinFreeSpace::Percent(0.0),
Vec::new(),
)
.unwrap();
// Prune ran as part of add_location's epilogue. Confirm the
// partial EC was left alone because the sibling .dat is too
// small to be the encoding source.
let ec_loc = &store.locations[1];
let ev = ec_loc
.find_ec_volume(VolumeId(vid))
.expect("partial EC volume must remain mounted when the sibling .dat is smaller than the .vif source size");
assert_eq!(ev.shard_count(), 1);
let ec_base = ec_dir
.join(format!("{}_{}", collection, vid))
.to_string_lossy()
.into_owned();
assert!(
std::path::Path::new(&format!("{}.ec01", ec_base)).exists(),
"partial shard file must survive when the sibling .dat is truncated",
);
assert!(
std::path::Path::new(&format!("{}.ecx", ec_base)).exists(),
".ecx must survive when the sibling .dat is truncated",
);
}
/// A distributed EC volume — no `.dat` on any disk of this store —
/// must not be touched by the prune even when shard count is below
/// `DATA_SHARDS_COUNT`. Partial shard layouts are normal for

View File

@ -288,6 +288,15 @@ func (ev *EcVolume) ShardSize() uint64 {
return 0
}
// DatFileSize returns the source .dat file size as recorded in .vif at
// EC encoding time. Zero for old EC volumes whose .vif predates the
// field, or for .vif files we failed to parse. Used by the Store-level
// prune in store_ec_reconcile.go to validate that a sibling-disk .dat
// is plausibly the encoding source before deleting the partial EC.
func (ev *EcVolume) DatFileSize() int64 {
return ev.datFileSize
}
func (ev *EcVolume) Size() (size uint64) {
for _, shard := range ev.Shards {
if shardSize := shard.Size(); shardSize > 0 {

View File

@ -163,3 +163,95 @@ func TestIssue9478_PartialEcOnSiblingDiskOfHealthyDat(t *testing.T) {
t.Fatalf("healthy .dat on sibling disk was removed by the prune; only the partial EC files should be cleaned up")
}
}
// TestIssue9478_ZeroByteSiblingDatKeepsPartialEc is the safety guard for
// pruneIncompleteEcWithSiblingDat: when the sibling .dat is zero bytes
// (or smaller than what .vif recorded as the encoding source), we'd
// rather keep the partial EC than delete it based on garbage. The
// partial shard may still combine with shards on other servers in a
// recoverable distributed-EC layout.
func TestIssue9478_ZeroByteSiblingDatKeepsPartialEc(t *testing.T) {
collection := ""
vid := needle.VolumeId(122)
root := t.TempDir()
datDir := root + "/sdd"
ecDir := root + "/sdf"
if err := os.MkdirAll(datDir, 0o755); err != nil {
t.Fatalf("mkdir datDir: %v", err)
}
if err := os.MkdirAll(ecDir, 0o755); err != nil {
t.Fatalf("mkdir ecDir: %v", err)
}
// Sibling .dat is a zero-byte shell — the kind volume servers 3
// and 5 in the issue 9478 report have.
datBase := erasure_coding.EcShardFileName(collection, datDir, int(vid))
if f, err := os.Create(datBase + ".dat"); err == nil {
f.Close()
} else {
t.Fatalf("create zero-byte dat: %v", err)
}
ecBase := erasure_coding.EcShardFileName(collection, ecDir, int(vid))
datFileSizeForShard := int64(10 * 1024 * 1024)
expectedShardSize := calculateExpectedShardSize(datFileSizeForShard, erasure_coding.DataShardsCount)
if f, err := os.Create(ecBase + erasure_coding.ToExt(1)); err == nil {
if err := f.Truncate(expectedShardSize); err != nil {
t.Fatalf("truncate shard: %v", err)
}
f.Close()
}
if f, err := os.Create(ecBase + ".ecx"); err == nil {
f.WriteString("dummy ecx")
f.Close()
}
if f, err := os.Create(ecBase + ".ecj"); err == nil {
f.Close()
}
if f, err := os.Create(ecBase + ".vif"); err == nil {
f.Close()
}
minFreeSpace := util.MinFreeSpace{Type: util.AsPercent, Percent: 1, Raw: "1"}
makeDisk := func(dir string) *DiskLocation {
dl := &DiskLocation{
Directory: dir,
DirectoryUuid: "test-uuid-" + dir,
IdxDirectory: dir,
DiskType: types.HddType,
MinFreeSpace: minFreeSpace,
}
dl.volumes = make(map[needle.VolumeId]*Volume)
dl.ecVolumes = make(map[needle.VolumeId]*erasure_coding.EcVolume)
return dl
}
datLoc := makeDisk(datDir)
ecLoc := makeDisk(ecDir)
store := &Store{
Locations: []*DiskLocation{datLoc, ecLoc},
NewEcShardsChan: make(chan master_pb.VolumeEcShardInformationMessage, 16),
DeletedEcShardsChan: make(chan master_pb.VolumeEcShardInformationMessage, 16),
}
if err := ecLoc.loadAllEcShards(nil); err != nil {
t.Logf("loadAllEcShards on ecDir returned: %v", err)
}
t.Cleanup(func() { closeEcVolumes(ecLoc) })
if ecLoc.EcShardCount() == 0 {
t.Fatalf("test setup is not reproducing the layout: per-disk EC load did not mount the partial shard")
}
store.pruneIncompleteEcWithSiblingDat()
if ecLoc.EcShardCount() == 0 {
t.Fatalf("prune deleted partial EC based on a zero-byte sibling .dat; the shard should have been left in place so distributed reconstruction is still possible")
}
if !util.FileExists(ecBase + erasure_coding.ToExt(1)) {
t.Fatalf("prune deleted the partial shard file based on a zero-byte sibling .dat")
}
if len(store.DeletedEcShardsChan) != 0 {
t.Errorf("prune pushed a DeletedEcShardsChan message despite skipping cleanup; nothing was actually deleted")
}
}

View File

@ -10,8 +10,19 @@ import (
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
)
// datOwnerInfo records both the disk that holds a .dat for a given
// (collection, vid) and the size on disk. The size is consulted by
// pruneIncompleteEcWithSiblingDat before deleting any EC artefacts:
// a zero-byte or truncated .dat is not a credible fallback, and we'd
// rather leave the partial EC in place than wipe it based on garbage.
type datOwnerInfo struct {
location *DiskLocation
size int64
}
// ecKeyForReconcile keys orphan-shard reconciliation by collection + volume
// id. Per-collection grouping matters because two collections can re-use the
// same volume id, and we must only pair shards with their own .ecx file.
@ -149,6 +160,14 @@ func (s *Store) indexEcxOwners() map[ecKeyForReconcile]ecxOwnerInfo {
// also fall through unchanged because the lookup in the .dat index below
// will simply not find a match.
//
// Before deleting any EC files we also check that the sibling .dat is
// plausibly the encoding source: at least super_block.SuperBlockSize
// bytes long, and — when the EC's .vif recorded a non-zero source size
// in datFileSize — at least that many bytes. A zero-byte shell or a
// truncated .dat does not justify wiping the partial EC, because that
// EC shard may still combine usefully with shards on other servers in
// a recoverable distributed-EC layout.
//
// We push DeletedEcShardsChan for every pruned shard so the master is told
// to forget the registrations the per-disk pass already emitted on
// NewEcShardsChan during startup, instead of waiting for the first
@ -181,15 +200,28 @@ func (s *Store) pruneIncompleteEcWithSiblingDat() {
continue
}
key := ecKeyForReconcile{collection: ev.Collection, vid: vid}
datOwner, hasDat := datOwners[key]
if !hasDat || datOwner == loc {
owner, hasDat := datOwners[key]
if !hasDat || owner.location == loc {
continue
}
// Decide whether the sibling .dat is a credible source.
// Prefer the size baked into .vif at encode time; fall
// back to "at least a superblock" for old EC volumes
// whose .vif predates the field.
requiredDatSize := ev.DatFileSize()
if requiredDatSize <= 0 {
requiredDatSize = int64(super_block.SuperBlockSize)
}
if owner.size < requiredDatSize {
glog.Warningf("ec volume %d (collection=%q) on %s has only %d shards but sibling .dat on %s is %d bytes (need >= %d); leaving partial EC in place so distributed reconstruction is still possible",
vid, ev.Collection, loc.Directory, shardCount, owner.location.Directory, owner.size, requiredDatSize)
continue
}
victims = append(victims, victim{
collection: ev.Collection,
vid: vid,
messages: ev.ToVolumeEcShardInformationMessage(uint32(diskId)),
datDir: datOwner.Directory,
datDir: owner.location.Directory,
shardCount: shardCount,
})
}
@ -215,18 +247,19 @@ func (s *Store) pruneIncompleteEcWithSiblingDat() {
}
// indexDatOwners returns, for every (collection, vid), the first disk on
// this store that holds a .dat file for it. Used by
// this store that holds a .dat file for it plus the file's size. Used by
// pruneIncompleteEcWithSiblingDat so it can decide whether partial EC
// artefacts on another disk are leftovers of an interrupted encode.
// artefacts on another disk are leftovers of an interrupted encode AND
// whether the sibling .dat is large enough to be a credible fallback.
//
// We accept any .dat that os.Stat can see — including the zero-byte
// shells the issue 9478 reporter shows on volume servers 3 and 5 — for
// the same reason checkDatFileExists in disk_location_ec.go does: the
// mere presence of a .dat means this volume was a regular volume on this
// server at some point, which rules out the "distributed EC, no .dat
// anywhere" reading that would justify keeping the partial shards.
func (s *Store) indexDatOwners() map[ecKeyForReconcile]*DiskLocation {
owners := make(map[ecKeyForReconcile]*DiskLocation)
// We record any .dat os.ReadDir can see — including zero-byte shells.
// The mere presence of a .dat means this volume was a regular volume on
// this server at some point, which rules out the "distributed EC, no
// .dat anywhere" reading. Whether that .dat is actually usable is the
// caller's call, made by comparing this size to the EC's recorded
// source size in .vif.
func (s *Store) indexDatOwners() map[ecKeyForReconcile]datOwnerInfo {
owners := make(map[ecKeyForReconcile]datOwnerInfo)
for _, loc := range s.Locations {
entries, err := os.ReadDir(loc.Directory)
if err != nil {
@ -245,9 +278,13 @@ func (s *Store) indexDatOwners() map[ecKeyForReconcile]*DiskLocation {
if err != nil {
continue
}
info, err := entry.Info()
if err != nil {
continue
}
key := ecKeyForReconcile{collection: collection, vid: vid}
if _, exists := owners[key]; !exists {
owners[key] = loc
owners[key] = datOwnerInfo{location: loc, size: info.Size()}
}
}
}