mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-16 12:16:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(master): advance maxVolumeId when registering EC shards (#9827)

Browse Source 
* fix(master): advance maxVolumeId when registering EC shards

After EC encoding the original normal volume is deleted, so a
high-numbered volume can exist only as EC shards. Only regular volumes
advanced maxVolumeId (Disk.doAddOrUpdateVolume), so a master that
rebuilt its state from heartbeats (raft state not resumed) undercounted
the max and NextVolumeId could re-issue an id that EC shards still
occupy. A new volume then gets created on top of the EC volume id; new
writes land on it, but reads route to the old EC shards whose .ecx never
held the new needle, returning 404 and corrupting that object.

Advance maxVolumeId when EC shards are registered, mirroring the
regular-volume path. RegisterEcShards is the chokepoint both the full
and incremental heartbeat sync paths funnel through.

* test: cover incremental heartbeat path for EC maxVolumeId

Both SyncDataNodeEcShards and IncrementalSyncDataNodeEcShards funnel
through RegisterEcShards; assert the invariant on the incremental path
too.
This commit is contained in:
Chris Lu
2026-06-04 22:25:30 -07:00
committed by GitHub
parent 8d59069a0a
commit 0d72023fac
2 changed files with 54 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/topology/topology_ec.go
+4
View File
@@ -129,6 +129,10 @@ func (loc *EcShardLocations) DeleteShard(shardId erasure_coding.ShardId, dn *Dat
func (t *Topology) RegisterEcShards(ecvi *erasure_coding.EcVolumeInfo, dn *DataNode) {
// EC-only volumes (source volume deleted after encoding) must bump
// maxVolumeId too, or a heartbeat-rebuilt master could re-issue their id.
t.UpAdjustMaxVolumeId(ecvi.VolumeId)
t.ecShardMapLock.Lock()
defer t.ecShardMapLock.Unlock()
weed/topology/topology_ec_max_volume_id_test.go
+50
View File
@@ -0,0 +1,50 @@
package topology
import (
"testing"
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
"github.com/seaweedfs/seaweedfs/weed/sequence"
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
)
// TestRegisterEcShardsAdvancesMaxVolumeId guards against volume-id reuse: an
// EC-only volume must advance maxVolumeId so a heartbeat-rebuilt master cannot
// re-issue its id.
func TestRegisterEcShardsAdvancesMaxVolumeId(t *testing.T) {
topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
dc := topo.GetOrCreateDataCenter("dc1")
rack := dc.GetOrCreateRack("rack1")
dn := rack.GetOrCreateDataNode("127.0.0.1", 34534, 0, "127.0.0.1", "", map[string]uint32{"": 100})
const ecVid = uint32(39848)
if got := topo.GetMaxVolumeId(); uint32(got) >= ecVid {
t.Fatalf("precondition: maxVolumeId already %d, want < %d", got, ecVid)
}
// Full-sync heartbeat reporting an EC-only volume.
msg := buildEcShardMessage(ecVid, "prefeitura", "", 0, []erasure_coding.ShardId{0, 1, 2, 3})
topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{msg}, dn)
if got := topo.GetMaxVolumeId(); got != needle.VolumeId(ecVid) {
t.Fatalf("maxVolumeId after EC registration = %d, want %d", got, ecVid)
}
}
// TestIncrementalSyncDataNodeEcShardsAdvancesMaxVolumeId locks the same
// invariant on the incremental heartbeat path.
func TestIncrementalSyncDataNodeEcShardsAdvancesMaxVolumeId(t *testing.T) {
topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
dc := topo.GetOrCreateDataCenter("dc1")
rack := dc.GetOrCreateRack("rack1")
dn := rack.GetOrCreateDataNode("127.0.0.1", 34534, 0, "127.0.0.1", "", map[string]uint32{"": 100})
const ecVid = uint32(39849)
msg := buildEcShardMessage(ecVid, "prefeitura", "", 0, []erasure_coding.ShardId{0, 1, 2, 3})
topo.IncrementalSyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{msg}, nil, dn)
if got := topo.GetMaxVolumeId(); got != needle.VolumeId(ecVid) {
t.Fatalf("maxVolumeId after incremental EC registration = %d, want %d", got, ecVid)
}
}