fix(ec): stamp each encode and reject cross-run shard reads

Generate stamps EncodeTsNs into the volume's .vif. Reads carry it to the
shard's owning volume (resolved together via FindEcVolumeWithShard, so a
multi-disk server validates the disk that actually serves the bytes) and
reject a shard from a different encode run, recovering from parity. A
zero on either side (pre-upgrade volume) skips the guard.

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))

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 {

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)
+	}
+}

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)

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
 				}