diff --git a/weed/storage/erasure_coding/ec_volume_test.go b/weed/storage/erasure_coding/ec_volume_test.go index 639a4c157..e338bc0b7 100644 --- a/weed/storage/erasure_coding/ec_volume_test.go +++ b/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) + } +} diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index ac1c5d678..aa2bd2318 100644 --- a/weed/storage/store_ec.go +++ b/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) }