diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs index 6c8485e08..b39368a38 100644 --- a/seaweed-volume/src/server/store_ec.rs +++ b/seaweed-volume/src/server/store_ec.rs @@ -35,6 +35,7 @@ use std::time::{Duration, Instant}; use futures::future::join_all; use futures::stream::{self, StreamExt}; use reed_solomon_erasure::galois_8::ReedSolomon; +use tokio::sync::Semaphore; use tonic::Request; use crate::pb::master_pb::{self, seaweed_client::SeaweedClient, LookupEcVolumeRequest}; @@ -54,6 +55,15 @@ use crate::storage::volume::volume_file_name; /// `ecIntervalReadConcurrency`. const INTERVAL_READ_CONCURRENCY: usize = 8; +/// Bounds the bytes EC recovery holds in flight across every concurrent read. +/// Recovery is the one read path that multiplies the served bytes — it keeps an +/// interval-sized buffer per shard alive until Reed-Solomon runs — and a peer +/// that is slow to fail holds each of them for the whole gRPC timeout, so a +/// burst of reads during a network blip walked the server into an OOM. Mirrors +/// Go's `ecRecoverBudget`. +const EC_RECOVER_BUDGET: usize = 256 << 20; +static EC_RECOVER_SEM: Semaphore = Semaphore::const_new(EC_RECOVER_BUDGET); + /// One interval's data after Phase A. enum IntervalResult { /// Already read from a locally-mounted shard. @@ -754,7 +764,7 @@ async fn fetch_one_interval( // Reconstruct: fan-out reads to every other shard at the same // (shard_offset, size). Mirrors `recoverOneRemoteEcShardInterval`. - let buf = recover_one_remote_ec_shard_interval( + recover_one_remote_ec_shard_interval( state, vid, needle_id, @@ -766,8 +776,7 @@ async fn fetch_one_interval( parity_shards, expected_encode_ts_ns, ) - .await?; - Ok((buf, false)) + .await } async fn read_remote_ec_shard_interval( @@ -927,7 +936,7 @@ async fn recover_one_remote_ec_shard_interval( data_shards: usize, parity_shards: usize, expected_encode_ts_ns: i64, -) -> io::Result> { +) -> io::Result<(Vec, bool)> { let total_shards = data_shards + parity_shards; let rs = ReedSolomon::new(data_shards, parity_shards).map_err(|e| { io::Error::new( @@ -936,89 +945,135 @@ async fn recover_one_remote_ec_shard_interval( ) })?; + // Charge the buffers this recovery is about to hold against the budget, so a + // burst of them queues here rather than on the heap. An interval whose + // fan-out outgrows the whole budget takes all of it and so runs alone, + // rather than waiting on permits that can never be granted. + let _permit = EC_RECOVER_SEM + .acquire_many((size * data_shards).min(EC_RECOVER_BUDGET) as u32) + .await + .map_err(|e| { + io::Error::new( + io::ErrorKind::Other, + format!( + "ec recover budget for shard {}.{}: {}", + vid.0, shard_id_to_recover, e + ), + ) + })?; + let mut bufs: Vec>> = vec![None; total_shards]; // Phase 0: seed bufs from LOCALLY mounted shards. If this node // already holds enough sibling shards, reconstruction completes // without any peer fan-out — and even with a cold/incomplete // shard_locations cache or a failed master lookup, local - // survivors still contribute. Mirrors Go's - // recoverOneRemoteEcShardInterval behaviour, which is implicitly - // local-aware because the Store fan-out targets ALL known - // locations (including the caller's own server address); the - // Rust port had been remote-only, so reconstructing with a cold - // cache failed even when enough siblings were on disk. + // survivors still contribute. + let mut available = 0usize; { let store = state.store.read().unwrap(); - if let Some(ecv) = store.find_ec_volume(vid) { - for sid in 0..total_shards { - if sid as ShardId == shard_id_to_recover { - continue; - } - if let Some(Some(shard)) = ecv.shards.get(sid) { - let mut buf = vec![0u8; size]; - if shard.read_at(&mut buf, shard_offset as u64).map(|n| n == size).unwrap_or(false) { - bufs[sid] = Some(buf); - } + for sid in 0..total_shards { + if available >= data_shards { + break; + } + if sid as ShardId == shard_id_to_recover { + continue; + } + // Resolve the shard together with the EcVolume on the disk that owns + // it: a reconciled volume has its shards split across data dirs. A + // shard from a different encode run must not be fed to Reed-Solomon; + // lenient only when the caller carries no identity (pre-upgrade). + // Mirrors Go's `readLocalEcShardInterval`. + let owner = match store.find_ec_volume_with_shard(vid, sid as u32) { + Some(ecv) if expected_encode_ts_ns == 0 || ecv.encode_ts_ns == expected_encode_ts_ns => ecv, + _ => continue, + }; + if let Some(Some(shard)) = owner.shards.get(sid) { + let mut buf = vec![0u8; size]; + if shard.read_at(&mut buf, shard_offset as u64).map(|n| n == size).unwrap_or(false) { + bufs[sid] = Some(buf); + available += 1; } } } } - // Phase 1: remote fan-out — one task per known shard location - // we DON'T already have locally and DON'T need to recover. - let mut tasks = Vec::new(); - for (sid, locs) in shard_locations { - if *sid == shard_id_to_recover || locs.is_empty() { - continue; - } - if bufs[*sid as usize].is_some() { - continue; - } - let sid = *sid; - let locs = locs.clone(); - let state = state.clone(); - tasks.push(async move { - let res = read_remote_ec_shard_interval( - &state, - &locs, - vid, - needle_id, - sid, - shard_offset, - size, - expected_encode_ts_ns, - ) - .await; - (sid, res) - }); - } - let results = join_all(tasks).await; + // Phase 1: remote fan-out over the shard locations we DON'T already have + // locally and DON'T need to recover. Reconstruction consumes data_shards + // shards, so reading every remaining one holds a third more buffers than + // that and asks a third more of peers that may already be struggling: fetch + // what is still missing, and widen only if some of those reads fail. + let mut candidates: Vec<(ShardId, Vec)> = shard_locations + .iter() + .filter(|(sid, locs)| { + **sid != shard_id_to_recover + && (**sid as usize) < total_shards + && !locs.is_empty() + && bufs[**sid as usize].is_none() + }) + .map(|(sid, locs)| (*sid, locs.clone())) + .collect(); - for (sid, res) in results { - match res { - // Exclude a deleted shard from reconstruction (Go gates on a full - // read): feeding the empty/zero buffer into Reed-Solomon would - // corrupt the recovered shard. - Ok((buf, is_deleted)) => { - if !is_deleted && (sid as usize) < total_shards { - bufs[sid as usize] = Some(buf); - } - } - Err(e) => { - tracing::debug!( - "recover: read {}.{} for needle {} failed: {}", - vid.0, - sid, + let mut any_deleted = false; + while available < data_shards && !candidates.is_empty() { + let rest = candidates.split_off((data_shards - available).min(candidates.len())); + let wave = std::mem::replace(&mut candidates, rest); + let results = join_all(wave.into_iter().map(|(sid, locs)| { + let state = state.clone(); + async move { + let res = read_remote_ec_shard_interval( + &state, + &locs, + vid, needle_id, - e - ); + sid, + shard_offset, + size, + expected_encode_ts_ns, + ) + .await; + (sid, res) } + })) + .await; + + for (sid, res) in results { + match res { + // Exclude a deleted shard from reconstruction (Go gates on a full + // read): feeding the empty/zero buffer into Reed-Solomon would + // corrupt the recovered shard. + Ok((buf, is_deleted)) => { + if is_deleted { + any_deleted = true; + continue; + } + bufs[sid as usize] = Some(buf); + available += 1; + } + Err(e) => { + tracing::debug!( + "recover: read {}.{} for needle {} failed: {}", + vid.0, + sid, + needle_id, + e + ); + } + } + } + if any_deleted { + // every shard of a deleted needle answers deleted, so another wave cannot help + break; } } - let available = bufs.iter().filter(|b| b.is_some()).count(); if available < data_shards { + // A holder reporting the needle deleted is authoritative -- deletes are + // never invented and never undone -- so answer that rather than the + // failure to gather shards of a needle that is gone. + if any_deleted { + return Ok((Vec::new(), true)); + } return Err(io::Error::new( io::ErrorKind::Other, format!( @@ -1039,7 +1094,7 @@ async fn recover_one_remote_ec_shard_interval( })?; match bufs.into_iter().nth(shard_id_to_recover as usize).flatten() { - Some(buf) => Ok(buf), + Some(buf) => Ok((buf, any_deleted)), None => Err(io::Error::new( io::ErrorKind::Other, format!( diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 85165330a..5ef495bba 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -12,6 +12,7 @@ import ( "time" "github.com/klauspost/reedsolomon" + "golang.org/x/sync/semaphore" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/operation" @@ -414,12 +415,15 @@ func (s *Store) ReadEcShardNeedle(vid needle.VolumeId, n *needle.Needle, onReadS glog.V(3).Infof("ReadEcShardNeedle needle id %s intervals:%+v", n.String(), intervals) } bytes, isDeleted, err := s.readEcShardIntervals(n.Id, localEcVolume, intervals) - if err != nil { - return 0, fmt.Errorf("ReadEcShardIntervals: %w", err) - } + // A holder reporting the needle deleted is authoritative -- deletes are + // never invented and never undone -- so answer that ahead of whatever + // error the shards it could not gather produced. if isDeleted { return 0, ErrorDeleted } + if err != nil { + return 0, fmt.Errorf("ReadEcShardIntervals: %w", err) + } err = n.ReadBytes(bytes, offset.ToActualOffset(), size, localEcVolume.Version) if err != nil { @@ -436,6 +440,15 @@ func (s *Store) IntervalToShardIdAndOffset(iv erasure_coding.Interval) (erasure_ return iv.ToShardIdAndOffset(erasure_coding.ErasureCodingLargeBlockSize, erasure_coding.ErasureCodingSmallBlockSize) } +var ( + // ecRecoverBudget bounds the interval-sized buffers EC recovery holds across + // all concurrent reads: a peer that is slow to fail keeps a whole fan-out of + // them alive for the gRPC timeout, and a burst of those walked servers into an + // OOM. A var so a test can narrow it alongside ecRecoverSem. + ecRecoverBudget int64 = 256 << 20 + ecRecoverSem = semaphore.NewWeighted(ecRecoverBudget) +) + // ecIntervalReadConcurrency bounds the fan-out of a single needle read. Blocks // that follow each other in the .dat live on different shards, so a needle // spanning several of them costs one round trip per block when read in sequence. @@ -713,47 +726,100 @@ func (s *Store) recoverOneRemoteEcShardInterval(needleId types.NeedleId, ecVolum return 0, false, fmt.Errorf("failed to create encoder: %w", err) } + // Charge the buffers this recovery is about to hold against the budget, so a + // burst of them queues here rather than on the heap. + weight := int64(len(buf)) * int64(ecCtx.DataShards) + if weight > ecRecoverBudget { + // An interval whose fan-out outgrows the whole budget takes all of it and + // so runs alone, rather than blocking forever on an acquire that can never + // succeed. The cap is then one such recovery, not a burst of them. + weight = ecRecoverBudget + } + if err = ecRecoverSem.Acquire(context.Background(), weight); err != nil { + return 0, false, err + } + defer ecRecoverSem.Release(weight) + // Use MaxShardCount to support custom EC ratios up to 32 shards 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 + // A shard this server already holds costs no round trip and no peer buffer, + // so seed those before asking peers for the rest. + available := 0 + for shardId := erasure_coding.ShardId(0); int(shardId) < ecCtx.Total() && available < ecCtx.DataShards; shardId++ { + if shardId == shardIdToRecover { + continue + } + if _, _, found := s.FindEcVolumeWithShard(ecVolume.VolumeId, shardId); !found { + continue + } + data := make([]byte, len(buf)) + if localErr := s.readLocalEcShardInterval(ecVolume, shardId, data, offset); localErr != nil { + glog.V(3).Infof("recover: read local ec shard %d.%d: %v", ecVolume.VolumeId, shardId, localErr) + continue + } + bufs[shardId] = data + available++ + } + + var candidates []erasure_coding.ShardId + candidateLocations := make(map[erasure_coding.ShardId][]pb.ServerAddress) ecVolume.ShardLocationsLock.RLock() for shardId, locations := range ecVolume.ShardLocations { - // skip current shard or empty shard - if shardId == shardIdToRecover { + // skip the shard being recovered, one already seeded locally, or an empty shard + if shardId == shardIdToRecover || int(shardId) >= ecCtx.Total() || bufs[shardId] != nil { continue } if len(locations) == 0 { glog.V(3).Infof("readRemoteEcShardInterval missing %d.%d from %+v", ecVolume.VolumeId, shardId, locations) continue } - - // read from remote locations - wg.Add(1) - 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, 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) - } - if isDeleted { - isDeletedFlag.Store(true) - } - if nRead == len(buf) { - bufs[shardId] = data - } - }(shardId, locations) + candidates = append(candidates, shardId) + candidateLocations[shardId] = locations } ecVolume.ShardLocationsLock.RUnlock() - wg.Wait() + // 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 + + // Reconstruction consumes DataShards buffers, so reading every remaining shard + // holds a third more than that and asks a third more of peers that are, by + // then, already struggling. Widen only when a wave falls short. + for len(candidates) > 0 && available < ecCtx.DataShards { + wave := min(ecCtx.DataShards-available, len(candidates)) + var wg sync.WaitGroup + var fetched atomic.Int64 + for _, shardId := range candidates[:wave] { + locations := candidateLocations[shardId] + wg.Add(1) + go func() { + defer wg.Done() + data := make([]byte, len(buf)) + 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) + } + if isDeleted { + isDeletedFlag.Store(true) + } + if nRead == len(buf) { + bufs[shardId] = data + fetched.Add(1) + } + }() + } + wg.Wait() + candidates = candidates[wave:] + available += int(fetched.Load()) + if isDeletedFlag.Load() { + // every shard of a deleted needle answers deleted, so another wave cannot help + break + } + } is_deleted = isDeletedFlag.Load() // Count and log available shards for diagnostics @@ -773,14 +839,14 @@ func (s *Store) recoverOneRemoteEcShardInterval(needleId types.NeedleId, ecVolum len(missingShards), missingShards) 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)", + return 0, is_deleted, fmt.Errorf("cannot recover shard %d.%d: only %d shards available %v, need at least %d (missing: %v)", ecVolume.VolumeId, shardIdToRecover, len(availableShards), availableShards, ecCtx.DataShards, missingShards) } 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", + return 0, is_deleted, fmt.Errorf("failed to reconstruct data for shard %d.%d with %d available shards %v: %w", ecVolume.VolumeId, shardIdToRecover, len(availableShards), availableShards, err) } glog.V(4).Infof("recovered ec shard %d.%d from other locations", ecVolume.VolumeId, shardIdToRecover) diff --git a/weed/storage/store_ec_recover_fanout_test.go b/weed/storage/store_ec_recover_fanout_test.go new file mode 100644 index 000000000..c30a3f443 --- /dev/null +++ b/weed/storage/store_ec_recover_fanout_test.go @@ -0,0 +1,359 @@ +package storage + +import ( + "bytes" + "math/rand" + "net" + "os" + "sync" + "sync/atomic" + "testing" + "time" + + "golang.org/x/sync/semaphore" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/storage/volume_info" +) + +// countingEcShardPeer stands in for a peer volume server that is having a +// transient problem: every VolumeEcShardRead is counted and then held until the +// test releases it, so a stalled fan-out is observable while it is stalled. +type countingEcShardPeer struct { + volume_server_pb.UnimplementedVolumeServerServer + requests atomic.Int64 + inFlight atomic.Int64 + peak atomic.Int64 + release chan struct{} + releaseOnce sync.Once +} + +func (p *countingEcShardPeer) VolumeEcShardRead(req *volume_server_pb.VolumeEcShardReadRequest, stream volume_server_pb.VolumeServer_VolumeEcShardReadServer) error { + p.requests.Add(1) + inFlight := p.inFlight.Add(1) + for peak := p.peak.Load(); inFlight > peak; peak = p.peak.Load() { + if p.peak.CompareAndSwap(peak, inFlight) { + break + } + } + defer p.inFlight.Add(-1) + <-p.release + return status.Error(codes.Unavailable, "transient failure") +} + +func (p *countingEcShardPeer) releaseAll() { + p.releaseOnce.Do(func() { close(p.release) }) +} + +func startCountingEcShardPeer(t *testing.T) (*countingEcShardPeer, pb.ServerAddress) { + t.Helper() + peer := &countingEcShardPeer{release: make(chan struct{})} + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + srv := grpc.NewServer() + volume_server_pb.RegisterVolumeServerServer(srv, peer) + go srv.Serve(lis) + t.Cleanup(func() { + peer.releaseAll() + srv.Stop() + }) + return peer, pb.NewServerAddressWithGrpcPort("127.0.0.1:1", lis.Addr().(*net.TCPAddr).Port) +} + +// writeEcVolumeFiles writes one needle into a volume in dir and EC-encodes it in +// place, leaving the .ec?? / .ecx / .vif set behind. +func writeEcVolumeFiles(t *testing.T, dir string, vid needle.VolumeId) (baseFileName string, n *needle.Needle) { + t.Helper() + + v, err := NewVolume(dir, dir, "", vid, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("new volume: %v", err) + } + n = new(needle.Needle) + n.Id = types.Uint64ToNeedleId(42) + n.Data = make([]byte, 3*erasure_coding.ErasureCodingSmallBlockSize+1234) + rand.New(rand.NewSource(42)).Read(n.Data) + n.Checksum = needle.NewCRC(n.Data) + if _, _, _, err := v.writeNeedle2(n, true, false, false); err != nil { + t.Fatalf("write needle: %v", err) + } + baseFileName = v.DataFileName() + v.Close() + + datSize, err := os.Stat(baseFileName + ".dat") + if err != nil { + t.Fatalf("stat .dat: %v", err) + } + if _, err := erasure_coding.WriteEcFiles(baseFileName, erasure_coding.BackgroundECContext()); err != nil { + t.Fatalf("write ec files: %v", err) + } + if err := erasure_coding.WriteSortedFileFromIdx(baseFileName, ".ecx"); err != nil { + t.Fatalf("write .ecx: %v", err) + } + if err := os.WriteFile(baseFileName+".ecj", nil, 0o644); err != nil { + t.Fatalf("write .ecj: %v", err) + } + if err := volume_info.SaveVolumeInfo(baseFileName+".vif", &volume_server_pb.VolumeInfo{ + Version: uint32(needle.GetCurrentVersion()), + DatFileSize: datSize.Size(), + EcShardConfig: &volume_server_pb.EcShardConfig{ + DataShards: erasure_coding.DataShardsCount, + ParityShards: erasure_coding.ParityShardsCount, + }, + }); err != nil { + t.Fatalf("save .vif: %v", err) + } + for _, ext := range []string{".dat", ".idx"} { + if err := os.Remove(baseFileName + ext); err != nil { + t.Fatalf("remove %s: %v", ext, err) + } + } + return baseFileName, n +} + +// mountLocalEcVolume writes an EC volume into the store's own disk and mounts +// every shard, so a recovery can be served entirely from local shards. +func mountLocalEcVolume(t *testing.T, store *Store, vid needle.VolumeId) (*erasure_coding.EcVolume, *needle.Needle) { + t.Helper() + _, n := writeEcVolumeFiles(t, store.Locations[0].Directory, vid) + for shardId := 0; shardId < erasure_coding.TotalShardsCount; shardId++ { + if err := store.MountEcShards("", vid, erasure_coding.ShardId(shardId), ""); err != nil { + t.Fatalf("mount shard %d: %v", shardId, err) + } + } + ecVolume, found := store.Locations[0].FindEcVolume(vid) + if !found { + t.Fatal("ec volume not mounted") + } + return ecVolume, n +} + +// seedShardLocations points every shard of the volume at one address and marks +// the cache fresh, so a read never reaches for the master this test does not run. +func seedShardLocations(ecVolume *erasure_coding.EcVolume, addr pb.ServerAddress) { + ecVolume.ShardLocationsLock.Lock() + for shardId := 0; shardId < erasure_coding.TotalShardsCount; shardId++ { + ecVolume.ShardLocations[erasure_coding.ShardId(shardId)] = []pb.ServerAddress{addr} + } + ecVolume.ShardLocationsRefreshTime = time.Now() + ecVolume.ShardLocationsLock.Unlock() +} + +// Recovery reconstructs a shard from the shards this server already holds, so a +// store holding the whole volume never asks a peer for one. +func TestRecoverOneRemoteEcShardIntervalUsesLocalShards(t *testing.T) { + store := newTestStore(t, 1) + peer, addr := startCountingEcShardPeer(t) + ecVolume, n := mountLocalEcVolume(t, store, 7) + seedShardLocations(ecVolume, addr) + + _, _, intervals, err := ecVolume.LocateEcShardNeedle(n.Id, ecVolume.Version) + if err != nil { + t.Fatalf("locate needle: %v", err) + } + shardIdToRecover, actualOffset := store.IntervalToShardIdAndOffset(intervals[0]) + + got := make([]byte, intervals[0].Size) + nRead, _, err := store.recoverOneRemoteEcShardInterval(n.Id, ecVolume, shardIdToRecover, got, actualOffset) + if err != nil { + t.Fatalf("recover ec shard interval: %v", err) + } + if nRead != len(got) { + t.Fatalf("recovered %d bytes, want %d", nRead, len(got)) + } + if requests := peer.requests.Load(); requests != 0 { + t.Errorf("recovery asked peers for %d shard intervals, want 0 with every shard local", requests) + } + + want := make([]byte, len(got)) + if err := store.readLocalEcShardInterval(ecVolume, shardIdToRecover, want, actualOffset); err != nil { + t.Fatalf("read local ec shard %d: %v", shardIdToRecover, err) + } + if !bytes.Equal(got, want) { + t.Errorf("recovered shard %d bytes differ from the shard on disk", shardIdToRecover) + } +} + +// shardServingPeer answers VolumeEcShardRead out of an EC volume's shard files +// and counts how many shard intervals the caller asked for. +type shardServingPeer struct { + volume_server_pb.UnimplementedVolumeServerServer + baseFileName string + requests atomic.Int64 +} + +func (p *shardServingPeer) VolumeEcShardRead(req *volume_server_pb.VolumeEcShardReadRequest, stream volume_server_pb.VolumeServer_VolumeEcShardReadServer) error { + p.requests.Add(1) + f, err := os.Open(p.baseFileName + erasure_coding.ToExt(int(req.ShardId))) + if err != nil { + return err + } + defer f.Close() + data := make([]byte, req.Size) + if _, err := f.ReadAt(data, req.Offset); err != nil { + return err + } + return stream.Send(&volume_server_pb.VolumeEcShardReadResponse{Data: data, EncodeTsNs: req.EncodeTsNs}) +} + +// Reed-Solomon needs DataShards shards, so the fan-out stops there instead of +// pulling every surviving shard off peers that may already be struggling. +func TestRecoverOneRemoteEcShardIntervalFetchesOnlyWhatItNeeds(t *testing.T) { + baseFileName, n := writeEcVolumeFiles(t, t.TempDir(), 7) + + peer := &shardServingPeer{baseFileName: baseFileName} + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + srv := grpc.NewServer() + volume_server_pb.RegisterVolumeServerServer(srv, peer) + go srv.Serve(lis) + t.Cleanup(srv.Stop) + addr := pb.NewServerAddressWithGrpcPort("127.0.0.1:1", lis.Addr().(*net.TCPAddr).Port) + + store := &Store{grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials())} + ecVolume := &erasure_coding.EcVolume{ + VolumeId: 7, + ShardLocations: make(map[erasure_coding.ShardId][]pb.ServerAddress), + } + seedShardLocations(ecVolume, addr) + + const shardIdToRecover = erasure_coding.ShardId(0) + const offset = 0 + got := make([]byte, 4096) + if _, _, err := store.recoverOneRemoteEcShardInterval(n.Id, ecVolume, shardIdToRecover, got, offset); err != nil { + t.Fatalf("recover ec shard interval: %v", err) + } + if requests := peer.requests.Load(); requests != int64(erasure_coding.DataShardsCount) { + t.Errorf("recovery asked peers for %d shard intervals, want %d", requests, erasure_coding.DataShardsCount) + } + + want := make([]byte, len(got)) + f, err := os.Open(baseFileName + erasure_coding.ToExt(int(shardIdToRecover))) + if err != nil { + t.Fatalf("open shard %d: %v", shardIdToRecover, err) + } + defer f.Close() + if _, err := f.ReadAt(want, offset); err != nil { + t.Fatalf("read shard %d: %v", shardIdToRecover, err) + } + if !bytes.Equal(got, want) { + t.Errorf("recovered shard %d bytes differ from the shard on disk", shardIdToRecover) + } +} + +// setEcRecoverBudget narrows the process-wide recovery budget for one test. +func setEcRecoverBudget(t *testing.T, budget int64) { + t.Helper() + oldBudget, oldSem := ecRecoverBudget, ecRecoverSem + ecRecoverBudget, ecRecoverSem = budget, semaphore.NewWeighted(budget) + t.Cleanup(func() { ecRecoverBudget, ecRecoverSem = oldBudget, oldSem }) +} + +// A burst of reads against a peer that has stopped answering queues on the +// recovery budget instead of piling interval-sized buffers onto the heap. +func TestRecoverOneRemoteEcShardIntervalBoundsInFlightBytes(t *testing.T) { + const intervalSize = 1 << 20 + const readers = 8 + const budget = 24 << 20 + + setEcRecoverBudget(t, budget) + peer, addr := startCountingEcShardPeer(t) + + store := &Store{grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials())} + ecVolume := &erasure_coding.EcVolume{ + VolumeId: 7, + ShardLocations: make(map[erasure_coding.ShardId][]pb.ServerAddress), + } + seedShardLocations(ecVolume, addr) + + var wg sync.WaitGroup + for i := 0; i < readers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + store.recoverOneRemoteEcShardInterval(types.Uint64ToNeedleId(42), ecVolume, 0, make([]byte, intervalSize), 0) + }() + } + + // let the fan-out reach the peer and settle there + for deadline, settled, last := time.Now().Add(10*time.Second), 0, int64(-1); time.Now().Before(deadline) && settled < 5; { + time.Sleep(50 * time.Millisecond) + inFlight := peer.inFlight.Load() + if inFlight > 0 && inFlight == last { + settled++ + } else { + settled = 0 + } + last = inFlight + } + + if peak, allowed := peer.peak.Load(), int64(budget/intervalSize); peak > allowed { + t.Errorf("%d shard reads of %dMB in flight at once, more than the %dMB budget allows", peak, intervalSize>>20, budget>>20) + } + + peer.releaseAll() + wg.Wait() +} + +// deletingEcShardPeer answers every shard read with the needle's deletion. +type deletingEcShardPeer struct { + volume_server_pb.UnimplementedVolumeServerServer +} + +func (p *deletingEcShardPeer) VolumeEcShardRead(req *volume_server_pb.VolumeEcShardReadRequest, stream volume_server_pb.VolumeServer_VolumeEcShardReadServer) error { + return stream.Send(&volume_server_pb.VolumeEcShardReadResponse{IsDeleted: true, EncodeTsNs: req.EncodeTsNs}) +} + +// A needle the peers report deleted reads as deleted -- a 404 -- even when too +// few shards came back to reconstruct it. +func TestReadEcShardNeedleAnswersDeletedWhenRecoveryFallsShort(t *testing.T) { + store := newTestStore(t, 1) + store.grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials()) + const vid = needle.VolumeId(7) + _, n := writeEcVolumeFiles(t, store.Locations[0].Directory, vid) + + // Too few local shards to reconstruct, and none for the shard the needle's + // first interval lands on, so that interval has to recover from peers. + for shardId := 1; shardId <= 5; shardId++ { + if err := store.MountEcShards("", vid, erasure_coding.ShardId(shardId), ""); err != nil { + t.Fatalf("mount shard %d: %v", shardId, err) + } + } + ecVolume, found := store.Locations[0].FindEcVolume(vid) + if !found { + t.Fatal("ec volume not mounted") + } + + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + srv := grpc.NewServer() + volume_server_pb.RegisterVolumeServerServer(srv, &deletingEcShardPeer{}) + go srv.Serve(lis) + t.Cleanup(srv.Stop) + + seedShardLocations(ecVolume, pb.NewServerAddressWithGrpcPort("127.0.0.1:1", lis.Addr().(*net.TCPAddr).Port)) + ecVolume.ShardLocationsLock.Lock() + delete(ecVolume.ShardLocations, erasure_coding.ShardId(0)) + ecVolume.ShardLocationsLock.Unlock() + + got := new(needle.Needle) + got.Id = n.Id + if _, err := store.ReadEcShardNeedle(vid, got, nil); err != ErrorDeleted { + t.Errorf("read a deleted needle returned %v, want %v", err, ErrorDeleted) + } +}