diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index b0b1ebc05..3cfb3b10a 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -141,10 +141,12 @@ func (vs *VolumeServer) VolumeEcShardsGenerate(ctx context.Context, req *volume_ ecCtx.DataShards, ecCtx.ParityShards, ecCtx.Total(), erasure_coding.MaxShardCount) } - // Save EC configuration to VolumeInfo + // EncodeTsNs stamps this run's identity into the .vif (copied with the + // shards), so a read served from a different run's shard is rejected. volumeInfo.EcShardConfig = &volume_server_pb.EcShardConfig{ DataShards: uint32(ecCtx.DataShards), ParityShards: uint32(ecCtx.ParityShards), + EncodeTsNs: time.Now().UnixNano(), } glog.V(1).Infof("Saving EC config to .vif for volume %d: %d+%d (total: %d)", req.VolumeId, ecCtx.DataShards, ecCtx.ParityShards, ecCtx.Total()) @@ -581,16 +583,19 @@ func (vs *VolumeServer) VolumeEcShardsUnmount(ctx context.Context, req *volume_s func (vs *VolumeServer) VolumeEcShardRead(req *volume_server_pb.VolumeEcShardReadRequest, stream volume_server_pb.VolumeServer_VolumeEcShardReadServer) error { - ecVolume, found := vs.store.FindEcVolume(needle.VolumeId(req.VolumeId)) - if !found { - return fmt.Errorf("VolumeEcShardRead not found ec volume id %d", req.VolumeId) - } - // shard may live on a sibling disk of this server; walk all of them - // under ecVolumesLock. - _, ecShard, found := vs.store.FindEcShard(needle.VolumeId(req.VolumeId), erasure_coding.ShardId(req.ShardId)) + // Resolve the shard together with the EcVolume on the disk that owns it, + // rather than a first-match volume on a sibling disk: on a multi-disk server + // those can belong to different encode generations, and the guard must + // validate the identity of the volume whose bytes we serve. + ecVolume, ecShard, found := vs.store.FindEcVolumeWithShard(needle.VolumeId(req.VolumeId), erasure_coding.ShardId(req.ShardId)) if !found { return fmt.Errorf("not found ec shard %d.%d", req.VolumeId, req.ShardId) } + // Reject a shard from a different encode run than the caller's index; the + // caller then recovers from parity. 0 on either side = pre-upgrade volume. + if req.EncodeTsNs != 0 && ecVolume.EncodeTsNs != 0 && req.EncodeTsNs != ecVolume.EncodeTsNs { + return fmt.Errorf("ec shard %d.%d belongs to a different encode run", req.VolumeId, req.ShardId) + } if req.FileKey != 0 { _, size, _ := ecVolume.FindNeedleFromEcx(types.Uint64ToNeedleId(req.FileKey)) diff --git a/weed/storage/erasure_coding/ec_volume.go b/weed/storage/erasure_coding/ec_volume.go index 4860ae2ce..89b3613d3 100644 --- a/weed/storage/erasure_coding/ec_volume.go +++ b/weed/storage/erasure_coding/ec_volume.go @@ -44,6 +44,10 @@ type EcVolume struct { ExpireAtSec uint64 //ec volume destroy time, calculated from the ec volume was created ECContext *ECContext // EC encoding parameters + // EncodeTsNs is the encode time (unix nanos) loaded from .vif; reads carry it + // so a shard from a different encode run is rejected. 0 for pre-upgrade volumes. + EncodeTsNs int64 + // ecjFileSize mirrors the on-disk size of the .ecj deletion journal and // is maintained under ecjFileAccessLock. It is only used by IO helpers // (seek/truncate) — the authoritative runtime delete count comes from @@ -150,6 +154,7 @@ func NewEcVolume(diskType types.DiskType, dir string, dirIdx string, collection if volumeInfo.EcShardConfig != nil { ds := int(volumeInfo.EcShardConfig.DataShards) ps := int(volumeInfo.EcShardConfig.ParityShards) + ev.EncodeTsNs = volumeInfo.EcShardConfig.GetEncodeTsNs() // Validate shard counts to prevent zero or invalid values if ds <= 0 || ps <= 0 || ds+ps > MaxShardCount { diff --git a/weed/storage/erasure_coding/ec_volume_test.go b/weed/storage/erasure_coding/ec_volume_test.go index 7d833502c..0864a2bd4 100644 --- a/weed/storage/erasure_coding/ec_volume_test.go +++ b/weed/storage/erasure_coding/ec_volume_test.go @@ -7,8 +7,10 @@ import ( "github.com/stretchr/testify/assert" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/storage/volume_info" ) func TestPositioning(t *testing.T) { @@ -52,3 +54,40 @@ func TestPositioning(t *testing.T) { } } + +// TestNewEcVolumeLoadsEncodeTsNs pins that the per-encode identity stamped into +// .vif is loaded onto the EcVolume, so reads can reject a shard from a different +// encode run. +func TestNewEcVolumeLoadsEncodeTsNs(t *testing.T) { + dir := t.TempDir() + const vid = needle.VolumeId(123) + base := EcShardFileName("", dir, int(vid)) + + // A 0-byte .ecx is a valid index (no live needles) and lets NewEcVolume mount. + if err := os.WriteFile(base+".ecx", nil, 0o644); err != nil { + t.Fatalf("write .ecx: %v", err) + } + + const tsNs int64 = 1717000000000000123 + vi := &volume_server_pb.VolumeInfo{ + Version: uint32(needle.Version3), + EcShardConfig: &volume_server_pb.EcShardConfig{ + DataShards: 10, + ParityShards: 4, + EncodeTsNs: tsNs, + }, + } + if err := volume_info.SaveVolumeInfo(base+".vif", vi); err != nil { + t.Fatalf("save .vif: %v", err) + } + + ev, err := NewEcVolume(types.HardDriveType, dir, dir, "", vid) + if err != nil { + t.Fatalf("NewEcVolume: %v", err) + } + defer ev.Close() + + if ev.EncodeTsNs != tsNs { + t.Errorf("EncodeTsNs = %d, want %d", ev.EncodeTsNs, tsNs) + } +} diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 018f18d72..3ef81fd77 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -294,6 +294,21 @@ func (s *Store) FindEcShard(vid needle.VolumeId, shardId erasure_coding.ShardId) return s.findEcShard(vid, shardId) } +// FindEcVolumeWithShard returns the EcVolume on the disk that owns the given +// shard, plus the shard. The read guard must check the identity of the volume +// that owns the bytes served: on a multi-disk server one vid can hold shards +// from different encode runs across disks, so a first-match volume can differ. +func (s *Store) FindEcVolumeWithShard(vid needle.VolumeId, shardId erasure_coding.ShardId) (*erasure_coding.EcVolume, *erasure_coding.EcVolumeShard, bool) { + for _, location := range s.Locations { + if shard, found := location.FindEcShard(vid, shardId); found { + if ev, ok := location.FindEcVolume(vid); ok { + return ev, shard, true + } + } + } + return nil, nil, false +} + func (s *Store) FindEcVolume(vid needle.VolumeId) (*erasure_coding.EcVolume, bool) { for _, location := range s.Locations { if s, found := location.FindEcVolume(vid); found { @@ -422,7 +437,7 @@ func (s *Store) readOneEcShardInterval(needleId types.NeedleId, ecVolume *erasur // try reading directly if hasShardIdLocation { - _, is_deleted, err = s.readRemoteEcShardInterval(sourceDataNodes, needleId, ecVolume.VolumeId, shardId, data, actualOffset) + _, is_deleted, err = s.readRemoteEcShardInterval(sourceDataNodes, needleId, ecVolume.VolumeId, shardId, data, actualOffset, ecVolume.EncodeTsNs) if err == nil { return } @@ -490,12 +505,17 @@ func (s *Store) cachedLookupEcShardLocations(ecVolume *erasure_coding.EcVolume) } func (s *Store) readLocalEcShardInterval(ecVolume *erasure_coding.EcVolume, shardId erasure_coding.ShardId, buf []byte, offset int64) error { - // findEcShard walks every DiskLocation under ecVolumesLock; the + // Resolve the shard together with the EcVolume on the disk that owns it; the // shard may live on a sibling disk of this server. - _, shard, found := s.findEcShard(ecVolume.VolumeId, shardId) + ownerVolume, shard, found := s.FindEcVolumeWithShard(ecVolume.VolumeId, shardId) if !found { return fmt.Errorf("shard %d for volume %d: %w", shardId, ecVolume.VolumeId, errShardNotLocal) } + // Skip a local shard from a different encode run than the caller's index; + // treat it as not-local so the read recovers from the correct generation. + if ecVolume.EncodeTsNs != 0 && ownerVolume.EncodeTsNs != 0 && ecVolume.EncodeTsNs != ownerVolume.EncodeTsNs { + return fmt.Errorf("shard %d for volume %d: %w", shardId, ecVolume.VolumeId, errShardNotLocal) + } readBytes, err := shard.ReadAt(buf, offset) if err != nil { @@ -508,7 +528,7 @@ func (s *Store) readLocalEcShardInterval(ecVolume *erasure_coding.EcVolume, shar return nil } -func (s *Store) readRemoteEcShardInterval(sourceDataNodes []pb.ServerAddress, needleId types.NeedleId, vid needle.VolumeId, shardId erasure_coding.ShardId, buf []byte, offset int64) (n int, is_deleted bool, err error) { +func (s *Store) readRemoteEcShardInterval(sourceDataNodes []pb.ServerAddress, needleId types.NeedleId, vid needle.VolumeId, shardId erasure_coding.ShardId, buf []byte, offset int64, expectedEncodeTsNs int64) (n int, is_deleted bool, err error) { if len(sourceDataNodes) == 0 { return 0, false, fmt.Errorf("failed to find ec shard %d.%d", vid, shardId) @@ -516,7 +536,7 @@ func (s *Store) readRemoteEcShardInterval(sourceDataNodes []pb.ServerAddress, ne for _, sourceDataNode := range sourceDataNodes { glog.V(3).Infof("read remote ec shard %d.%d from %s", vid, shardId, sourceDataNode) - n, is_deleted, err = s.doReadRemoteEcShardInterval(sourceDataNode, needleId, vid, shardId, buf, offset) + n, is_deleted, err = s.doReadRemoteEcShardInterval(sourceDataNode, needleId, vid, shardId, buf, offset, expectedEncodeTsNs) if err == nil { return } @@ -526,17 +546,18 @@ func (s *Store) readRemoteEcShardInterval(sourceDataNodes []pb.ServerAddress, ne return } -func (s *Store) doReadRemoteEcShardInterval(sourceDataNode pb.ServerAddress, needleId types.NeedleId, vid needle.VolumeId, shardId erasure_coding.ShardId, buf []byte, offset int64) (n int, is_deleted bool, err error) { +func (s *Store) doReadRemoteEcShardInterval(sourceDataNode pb.ServerAddress, needleId types.NeedleId, vid needle.VolumeId, shardId erasure_coding.ShardId, buf []byte, offset int64, expectedEncodeTsNs int64) (n int, is_deleted bool, err error) { err = operation.WithVolumeServerClient(false, sourceDataNode, s.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { // copy data slice shardReadClient, err := client.VolumeEcShardRead(context.Background(), &volume_server_pb.VolumeEcShardReadRequest{ - VolumeId: uint32(vid), - ShardId: uint32(shardId), - Offset: offset, - Size: int64(len(buf)), - FileKey: uint64(needleId), + VolumeId: uint32(vid), + ShardId: uint32(shardId), + Offset: offset, + Size: int64(len(buf)), + FileKey: uint64(needleId), + EncodeTsNs: expectedEncodeTsNs, }) if err != nil { return fmt.Errorf("failed to start reading ec shard %d.%d from %s: %v", vid, shardId, sourceDataNode, err) @@ -595,7 +616,7 @@ func (s *Store) recoverOneRemoteEcShardInterval(needleId types.NeedleId, ecVolum go func(shardId erasure_coding.ShardId, locations []pb.ServerAddress) { defer wg.Done() data := make([]byte, len(buf)) - nRead, isDeleted, readErr := s.readRemoteEcShardInterval(locations, needleId, ecVolume.VolumeId, shardId, data, offset) + nRead, isDeleted, readErr := s.readRemoteEcShardInterval(locations, needleId, ecVolume.VolumeId, shardId, data, offset, ecVolume.EncodeTsNs) if readErr != nil { glog.V(3).Infof("recover: readRemoteEcShardInterval %d.%d %d bytes from %+v: %v", ecVolume.VolumeId, shardId, nRead, locations, readErr) forgetShardId(ecVolume, shardId) diff --git a/weed/storage/store_ec_scrub.go b/weed/storage/store_ec_scrub.go index 7302b2876..fbd9eccce 100644 --- a/weed/storage/store_ec_scrub.go +++ b/weed/storage/store_ec_scrub.go @@ -54,7 +54,7 @@ func (s *Store) ScrubEcVolume(vid needle.VolumeId) (int64, []*volume_server_pb.E sourceDataNodes, ok := ecv.ShardLocations[shardId] ecv.ShardLocationsLock.RUnlock() if ok { - if _, _, err := s.readRemoteEcShardInterval(sourceDataNodes, id, ecv.VolumeId, shardId, chunk, offset); err == nil { + if _, _, err := s.readRemoteEcShardInterval(sourceDataNodes, id, ecv.VolumeId, shardId, chunk, offset, ecv.EncodeTsNs); err == nil { data = append(data, chunk...) continue }