fix(ec): recover EC shards with the volume's own ratio, not the build default (#9958)

* fix(ec): recover EC shards with the volume's own ratio, not the build default

recoverOneRemoteEcShardInterval rebuilt a missing shard with a hardcoded
10+4 Reed-Solomon matrix (and counted sufficiency / iterated shards
against the 10+4 constants). For a custom-ratio volume (e.g. 9+3) that
reconstructs with the wrong matrix and corrupts the recovered bytes, and
cachedLookupEcShardLocations could wrongly reject a degraded but
recoverable custom-ratio read. Use the volume's own ECContext (loaded
from its .vif) for the encoder, the shard-iteration bound, and the
data-shard sufficiency checks. In OSS the ratio is always 10+4 so this is
a no-op; it brings the Go volume server in line with the Rust one, which
already reconstructs with the volume's ratio.

* fix(ec): close data races in the EC read-recovery path

Address review: the freshness check in cachedLookupEcShardLocations read
ecVolume.ShardLocations / ShardLocationsRefreshTime without the lock while
recover goroutines mutate them via forgetShardId -- snapshot both under
ShardLocationsLock.RLock(). The recover goroutines also wrote the shared
is_deleted return concurrently -- collect it via an atomic and fold it in
after they join. Also size availableShards/missingShards by the volume's
ECContext ratio rather than the 10+4 constants.

weed/storage/erasure_coding/ec_volume_test.go

@@ -119,3 +119,39 @@ func TestNewEcVolumeDoesNotWriteStubVif(t *testing.T) {
 		t.Fatalf("expected default EC context, got %+v", ev.ECContext)
 	}
 }
+
+// TestNewEcVolumeLoadsCustomRatio pins that a volume's own EC ratio is loaded from
+// its .vif into ECContext, so the read/recover/decode paths reconstruct with the
+// matrix that produced the shards rather than the build default. (Custom ratios are
+// an enterprise feature; in OSS the .vif always records 10+4.)
+func TestNewEcVolumeLoadsCustomRatio(t *testing.T) {
+	dir := t.TempDir()
+	const vid = needle.VolumeId(125)
+	base := EcShardFileName("", dir, int(vid))
+
+	if err := os.WriteFile(base+".ecx", nil, 0o644); err != nil {
+		t.Fatalf("write .ecx: %v", err)
+	}
+	if err := volume_info.SaveVolumeInfo(base+".vif", &volume_server_pb.VolumeInfo{
+		Version: uint32(needle.Version3),
+		EcShardConfig: &volume_server_pb.EcShardConfig{
+			DataShards:   9,
+			ParityShards: 3,
+		},
+	}); 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.ECContext == nil {
+		t.Fatalf("ECContext must be set from the .vif")
+	}
+	if ev.ECContext.DataShards != 9 || ev.ECContext.ParityShards != 3 {
+		t.Fatalf("ECContext = %d+%d, want 9+3", ev.ECContext.DataShards, ev.ECContext.ParityShards)
+	}
+}

weed/storage/store_ec.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"slices"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/klauspost/reedsolomon"
@@ -495,13 +496,27 @@ func forgetShardId(ecVolume *erasure_coding.EcVolume, shardId erasure_coding.Sha
 
 func (s *Store) cachedLookupEcShardLocations(ecVolume *erasure_coding.EcVolume) (err error) {
 
+	// Use the volume's own EC ratio so a custom-ratio volume (e.g. 9+3) is judged
+	// complete/recoverable against its real data-shard count, not the build default.
+	// In OSS the ratio is always 10+4, so this is a no-op.
+	ecCtx := ecVolume.ECContext
+	if ecCtx == nil {
+		ecCtx = erasure_coding.NewDefaultECContext(ecVolume.Collection, ecVolume.VolumeId)
+	}
+
+	// Snapshot the shard map size and refresh time under the lock: recover
+	// goroutines mutate ShardLocations via forgetShardId, so an unguarded read here
+	// races with a concurrent map write.
+	ecVolume.ShardLocationsLock.RLock()
 	shardCount := len(ecVolume.ShardLocations)
-	if shardCount < erasure_coding.DataShardsCount &&
-		ecVolume.ShardLocationsRefreshTime.Add(11*time.Second).After(time.Now()) ||
-		shardCount == erasure_coding.TotalShardsCount &&
-			ecVolume.ShardLocationsRefreshTime.Add(37*time.Minute).After(time.Now()) ||
-		shardCount >= erasure_coding.DataShardsCount &&
-			ecVolume.ShardLocationsRefreshTime.Add(7*time.Minute).After(time.Now()) {
+	refreshTime := ecVolume.ShardLocationsRefreshTime
+	ecVolume.ShardLocationsLock.RUnlock()
+	if shardCount < ecCtx.DataShards &&
+		refreshTime.Add(11*time.Second).After(time.Now()) ||
+		shardCount == ecCtx.Total() &&
+			refreshTime.Add(37*time.Minute).After(time.Now()) ||
+		shardCount >= ecCtx.DataShards &&
+			refreshTime.Add(7*time.Minute).After(time.Now()) {
 		// still fresh
 		return nil
 	}
@@ -516,8 +531,8 @@ func (s *Store) cachedLookupEcShardLocations(ecVolume *erasure_coding.EcVolume)
 		if err != nil {
 			return fmt.Errorf("lookup ec volume %d: %v", ecVolume.VolumeId, err)
 		}
-		if len(resp.ShardIdLocations) < erasure_coding.DataShardsCount {
-			return fmt.Errorf("only %d shards found but %d required", len(resp.ShardIdLocations), erasure_coding.DataShardsCount)
+		if len(resp.ShardIdLocations) < ecCtx.DataShards {
+			return fmt.Errorf("only %d shards found but %d required", len(resp.ShardIdLocations), ecCtx.DataShards)
 		}
 
 		ecVolume.ShardLocationsLock.Lock()
@@ -641,7 +656,15 @@ func (s *Store) doReadRemoteEcShardInterval(sourceDataNode pb.ServerAddress, nee
 func (s *Store) recoverOneRemoteEcShardInterval(needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, shardIdToRecover erasure_coding.ShardId, buf []byte, offset int64) (n int, is_deleted bool, err error) {
 	glog.V(3).Infof("recover ec shard %d.%d from other locations", ecVolume.VolumeId, shardIdToRecover)
 
-	enc, err := reedsolomon.New(erasure_coding.DataShardsCount, erasure_coding.ParityShardsCount)
+	// Reconstruct with the volume's OWN EC ratio (loaded from its .vif), not the
+	// build default, so a custom-ratio volume (e.g. 9+3) is decoded with the matrix
+	// that actually produced its shards -- decoding it as 10+4 would corrupt the
+	// recovered bytes. In OSS the ratio is always 10+4, so this is a no-op.
+	ecCtx := ecVolume.ECContext
+	if ecCtx == nil {
+		ecCtx = erasure_coding.NewDefaultECContext(ecVolume.Collection, ecVolume.VolumeId)
+	}
+	enc, err := reedsolomon.New(ecCtx.DataShards, ecCtx.ParityShards)
 	if err != nil {
 		return 0, false, fmt.Errorf("failed to create encoder: %w", err)
 	}
@@ -650,6 +673,10 @@ func (s *Store) recoverOneRemoteEcShardInterval(needleId types.NeedleId, ecVolum
 	bufs := make([][]byte, erasure_coding.MaxShardCount)
 
 	var wg sync.WaitGroup
+	// The recover goroutines run concurrently, so the deleted flag is collected
+	// atomically and folded into the named return after they join, rather than each
+	// goroutine writing the shared bool directly.
+	var isDeletedFlag atomic.Bool
 	ecVolume.ShardLocationsLock.RLock()
 	for shardId, locations := range ecVolume.ShardLocations {
 
@@ -673,7 +700,7 @@ func (s *Store) recoverOneRemoteEcShardInterval(needleId types.NeedleId, ecVolum
 				forgetShardId(ecVolume, shardId)
 			}
 			if isDeleted {
-				is_deleted = true
+				isDeletedFlag.Store(true)
 			}
 			if nRead == len(buf) {
 				bufs[shardId] = data
@@ -683,11 +710,12 @@ func (s *Store) recoverOneRemoteEcShardInterval(needleId types.NeedleId, ecVolum
 	ecVolume.ShardLocationsLock.RUnlock()
 
 	wg.Wait()
+	is_deleted = isDeletedFlag.Load()
 
 	// Count and log available shards for diagnostics
-	availableShards := make([]erasure_coding.ShardId, 0, erasure_coding.TotalShardsCount)
-	missingShards := make([]erasure_coding.ShardId, 0, erasure_coding.ParityShardsCount+1)
-	for shardId := 0; shardId < erasure_coding.TotalShardsCount; shardId++ {
+	availableShards := make([]erasure_coding.ShardId, 0, ecCtx.Total())
+	missingShards := make([]erasure_coding.ShardId, 0, ecCtx.ParityShards+1)
+	for shardId := 0; shardId < ecCtx.Total(); shardId++ {
 		if bufs[shardId] != nil {
 			availableShards = append(availableShards, erasure_coding.ShardId(shardId))
 		} else {
@@ -700,14 +728,14 @@ func (s *Store) recoverOneRemoteEcShardInterval(needleId types.NeedleId, ecVolum
 		len(availableShards), availableShards,
 		len(missingShards), missingShards)
 
-	if len(availableShards) < erasure_coding.DataShardsCount {
+	if len(availableShards) < ecCtx.DataShards {
 		return 0, false, fmt.Errorf("cannot recover shard %d.%d: only %d shards available %v, need at least %d (missing: %v)",
 			ecVolume.VolumeId, shardIdToRecover,
 			len(availableShards), availableShards,
-			erasure_coding.DataShardsCount, missingShards)
+			ecCtx.DataShards, missingShards)
 	}
 
-	if err = enc.ReconstructData(bufs[:erasure_coding.TotalShardsCount]); err != nil {
+	if err = enc.ReconstructData(bufs[:ecCtx.Total()]); err != nil {
 		return 0, false, fmt.Errorf("failed to reconstruct data for shard %d.%d with %d available shards %v: %w",
 			ecVolume.VolumeId, shardIdToRecover, len(availableShards), availableShards, err)
 	}