From e23a952878be437132d5e099f6d55f7ee8dff6b6 Mon Sep 17 00:00:00 2001 From: Mansur Akhmadi Date: Thu, 12 Nov 2026 00:57:39 +0700 Subject: [PATCH] ec: don't index ev.Shards[0] when the last shard was unmounted (#12538) * ec: don't index ev.Shards[0] when the last shard was unmounted * ec: skip a runtime emptied before the scrub merge reads its shards FindAllEcVolumes releases each location lock before MergeEcRuntimes runs, so an unmount can leave a merged runtime with no shards and the width scan indexed Shards[-1]. Co-Authored-By: Claude Opus 5.5 * ec: treat a scrub merge with no shard left as a vanished volume Co-Authored-By: Claude Opus 5.5 * ec: read Shards once in ShardSize so a concurrent unmount cannot empty it Co-Authored-By: Claude Opus 5.5 * ec: read each runtime's Shards once in the scrub merge The vanished check and the slot construction read Shards separately, so an unmount between them produced a non-nil view with no shards. Co-Authored-By: Claude Opus 5.5 * ec: anchor the scrub merge on a generation that still holds shards An unmount can empty the newest runtime while an older one keeps its shards, which left a merged view anchored on the emptied generation with no slots and the holder reported as skipped. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Mansur Akhmadi <10739460+mansur-akhmadi@users.noreply.github.com> Co-authored-by: Chris Lu Co-authored-by: Claude Opus 5.5 --- weed/storage/erasure_coding/ec_volume.go | 17 +++-- .../storage/erasure_coding/ec_volume_merge.go | 27 ++++++-- .../erasure_coding/ec_volume_merge_test.go | 55 ++++++++++++++++ weed/storage/erasure_coding/ec_volume_test.go | 63 +++++++++++++++++++ 4 files changed, 152 insertions(+), 10 deletions(-) create mode 100644 weed/storage/erasure_coding/ec_volume_merge_test.go diff --git a/weed/storage/erasure_coding/ec_volume.go b/weed/storage/erasure_coding/ec_volume.go index c3b6f81311..a77ab21f85 100644 --- a/weed/storage/erasure_coding/ec_volume.go +++ b/weed/storage/erasure_coding/ec_volume.go @@ -624,8 +624,8 @@ func (ev *EcVolume) EcIndexBaseFileName() string { } func (ev *EcVolume) ShardSize() uint64 { - if len(ev.Shards) > 0 { - return uint64(ev.Shards[0].Size()) + if shards := ev.Shards; len(shards) > 0 { + return uint64(shards[0].Size()) } return 0 } @@ -1010,11 +1010,13 @@ func (ev *EcVolume) LocateEcShardNeedle(needleId types.NeedleId, version needle. } intervals = ev.LocateEcShardNeedleInterval(version, offset.ToActualOffset(), size) + if len(intervals) == 0 && !size.IsDeleted() { + return types.Offset{}, 0, nil, fmt.Errorf("ec volume %d has no mounted shard to locate needle %s", ev.VolumeId, needleId) + } return } func (ev *EcVolume) LocateEcShardNeedleInterval(version needle.Version, offset int64, size types.Size) (intervals []Interval) { - shard := ev.Shards[0] var shardSize int64 if ev.datFileSize > 0 { // Use datFileSize to calculate the shardSize to match the EC encoding logic. @@ -1025,7 +1027,14 @@ func (ev *EcVolume) LocateEcShardNeedleInterval(version needle.Version, offset i // Subtract 1 to handle the ambiguous case where ecdFileSize is an exact // multiple of ErasureCodingLargeBlockSize but the data is actually in small // blocks (e.g., datFileSize was just under DataShards*ErasureCodingLargeBlockSize). - shardSize = shard.ecdFileSize - 1 + // A reader can still hold this EcVolume after its last shard was + // unmounted, so the slice may be empty. Return no intervals then; + // LocateEcShardNeedle reports it as an error. + shards := ev.Shards + if len(shards) == 0 { + return nil + } + shardSize = shards[0].ecdFileSize - 1 } // calculate the locations in the ec shards intervals = LocateData(ev.ECContext.LargeBlockSize(), ev.ECContext.SmallBlockSize(), shardSize, offset, types.Size(needle.GetActualSize(size, version))) diff --git a/weed/storage/erasure_coding/ec_volume_merge.go b/weed/storage/erasure_coding/ec_volume_merge.go index d031264b45..953edf8714 100644 --- a/weed/storage/erasure_coding/ec_volume_merge.go +++ b/weed/storage/erasure_coding/ec_volume_merge.go @@ -35,14 +35,22 @@ type MergedEcRuntimes struct { } // MergeEcRuntimes resolves runtimes (one vid's mounts, in location order) into -// a merged view. Returns nil for an empty slice (the vanished-volume case). +// a merged view. Returns nil when no runtime holds a shard (the vanished-volume +// case). Each runtime's Shards is read once, and a runtime emptied by a +// concurrent unmount neither anchors the generation nor is reported skipped. func MergeEcRuntimes(runtimes []*EcVolume) *MergedEcRuntimes { - if len(runtimes) == 0 { + held := make(map[*EcVolume][]*EcVolumeShard, len(runtimes)) + for _, v := range runtimes { + if shards := v.Shards; len(shards) > 0 { + held[v] = shards + } + } + if len(held) == 0 { return nil } anchorGen := int64(0) - for _, v := range runtimes { + for v := range held { if v.EncodeTsNs > anchorGen { anchorGen = v.EncodeTsNs } @@ -57,7 +65,7 @@ func MergeEcRuntimes(runtimes []*EcVolume) *MergedEcRuntimes { anchor := genMatches[0] for _, v := range genMatches { - if len(v.Shards) > 0 { + if len(held[v]) > 0 { anchor = v break } @@ -73,13 +81,17 @@ func MergeEcRuntimes(runtimes []*EcVolume) *MergedEcRuntimes { total := ecDataShards(anchor) + ecParityShards(anchor) width := total for _, v := range merged { - if n := int(v.Shards[len(v.Shards)-1].ShardId) + 1; n > width { + shards := held[v] + if len(shards) == 0 { + continue + } + if n := int(shards[len(shards)-1].ShardId) + 1; n > width { width = n } } slots := make([]*EcVolumeShard, width) for _, v := range merged { - for _, shard := range v.Shards { + for _, shard := range held[v] { id := int(shard.ShardId) if id >= len(slots) { continue @@ -92,6 +104,9 @@ func MergeEcRuntimes(runtimes []*EcVolume) *MergedEcRuntimes { var skipped []string for pos, v := range runtimes { + if len(held[v]) == 0 { + continue + } if v.EncodeTsNs != anchorGen { skipped = append(skipped, fmt.Sprintf( "EC volume %d shards at %s (position %d) belong to encode run %d but the scrub anchors on %d; they were not verified", diff --git a/weed/storage/erasure_coding/ec_volume_merge_test.go b/weed/storage/erasure_coding/ec_volume_merge_test.go new file mode 100644 index 0000000000..518a099751 --- /dev/null +++ b/weed/storage/erasure_coding/ec_volume_merge_test.go @@ -0,0 +1,55 @@ +package erasure_coding + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +// FindAllEcVolumes returns runtimes whose last shard can be unmounted before +// the merge reads them. +func TestMergeEcRuntimesWithEmptiedRuntime(t *testing.T) { + newRuntime := func(shardIds ...ShardId) *EcVolume { + ev := &EcVolume{ + VolumeId: needle.VolumeId(7), + ECContext: NewDefaultECContext("", 7), + EncodeTsNs: 100, + } + for _, id := range shardIds { + ev.Shards = append(ev.Shards, &EcVolumeShard{VolumeId: 7, ShardId: id}) + } + return ev + } + emptied := newRuntime() + holder := newRuntime(2, 5) + + merged := MergeEcRuntimes([]*EcVolume{emptied, holder}) + require.NotNil(t, merged) + assert.Same(t, holder, merged.Anchor) + assert.Len(t, merged.Merged, 2) + assert.Same(t, holder.Shards[0], merged.Slots[2]) + assert.Same(t, holder.Shards[1], merged.Slots[5]) +} + +func TestMergeEcRuntimesAllEmptiedIsVanished(t *testing.T) { + emptied := func() *EcVolume { + return &EcVolume{VolumeId: 7, ECContext: NewDefaultECContext("", 7), EncodeTsNs: 100} + } + assert.Nil(t, MergeEcRuntimes([]*EcVolume{emptied(), emptied()})) +} + +func TestMergeEcRuntimesEmptiedNewerGenerationDoesNotAnchor(t *testing.T) { + newer := &EcVolume{VolumeId: 7, ECContext: NewDefaultECContext("", 7), EncodeTsNs: 200} + older := &EcVolume{VolumeId: 7, ECContext: NewDefaultECContext("", 7), EncodeTsNs: 100} + older.Shards = []*EcVolumeShard{{VolumeId: 7, ShardId: 3}} + + merged := MergeEcRuntimes([]*EcVolume{newer, older}) + require.NotNil(t, merged) + assert.Same(t, older, merged.Anchor) + assert.Equal(t, []*EcVolume{older}, merged.Merged) + assert.Same(t, older.Shards[0], merged.Slots[3]) + assert.Empty(t, merged.Skipped) +} diff --git a/weed/storage/erasure_coding/ec_volume_test.go b/weed/storage/erasure_coding/ec_volume_test.go index 556fc1abab..116835f100 100644 --- a/weed/storage/erasure_coding/ec_volume_test.go +++ b/weed/storage/erasure_coding/ec_volume_test.go @@ -7,6 +7,7 @@ import ( "testing" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/storage/needle" @@ -283,3 +284,65 @@ func TestLocateEcShardNeedleMatchesInterval(t *testing.T) { expectedIntervals := ev.LocateEcShardNeedleInterval(needle.GetCurrentVersion(), offset.ToActualOffset(), size) assert.Equal(t, expectedIntervals, intervals) } + +// A reader holds the EcVolume it got from FindEcVolume after the location lock +// is released. If the last shard is unmounted in that window (ec.balance, +// ec.decode, volumeServer.evacuate), ev.Shards is empty when the reader locates +// the needle; indexing it panicked the volume server. +func TestLocateEcShardNeedleIntervalAfterLastShardUnmounted(t *testing.T) { + const datFileSize = 10 * 1118830592 + version := needle.GetCurrentVersion() + offset, size := int64(5_000_000_000), types.Size(1167) + withShard := []*EcVolumeShard{{ecdFileSize: datFileSize / DataShardsCount}} + + newVolume := func(dat int64, shards []*EcVolumeShard) *EcVolume { + return &EcVolume{ + VolumeId: needle.VolumeId(389), + Shards: shards, + ECContext: NewDefaultECContext("", 389), + datFileSize: dat, + } + } + + // datFileSize from .vif does not need any mounted shard. + want := newVolume(datFileSize, withShard).LocateEcShardNeedleInterval(version, offset, size) + assert.NotEmpty(t, want) + assert.Equal(t, want, newVolume(datFileSize, nil).LocateEcShardNeedleInterval(version, offset, size)) + assert.Equal(t, want, newVolume(datFileSize, []*EcVolumeShard{}).LocateEcShardNeedleInterval(version, offset, size)) + + // Legacy volumes size the shards from a mounted one; with none left + // there is nothing to locate. + assert.NotEmpty(t, newVolume(0, withShard).LocateEcShardNeedleInterval(version, offset, size)) + assert.Empty(t, newVolume(0, nil).LocateEcShardNeedleInterval(version, offset, size)) + assert.Empty(t, newVolume(0, []*EcVolumeShard{}).LocateEcShardNeedleInterval(version, offset, size)) +} + +// The read and delete paths go through LocateEcShardNeedle. Without a mounted +// shard to size a legacy volume they must get an error, not empty intervals +// that the read path would parse as a needle header. +func TestLocateEcShardNeedleWithoutMountedShards(t *testing.T) { + ecxFile, err := os.OpenFile("./test_files/389.ecx", os.O_RDONLY, 0) + require.NoError(t, err) + defer ecxFile.Close() + stat, err := ecxFile.Stat() + require.NoError(t, err) + + needleId, _ := types.ParseNeedleId("0f0edb92") + newVolume := func(dat int64) *EcVolume { + return &EcVolume{ + VolumeId: needle.VolumeId(389), + ecxFile: ecxFile, + ecxFileSize: stat.Size(), + ECContext: NewDefaultECContext("", 389), + datFileSize: dat, + } + } + + _, _, intervals, err := newVolume(0).LocateEcShardNeedle(needleId, needle.GetCurrentVersion()) + assert.ErrorContains(t, err, "no mounted shard") + assert.Empty(t, intervals) + + _, _, intervals, err = newVolume(10*1118830592).LocateEcShardNeedle(needleId, needle.GetCurrentVersion()) + assert.NoError(t, err) + assert.NotEmpty(t, intervals) +}