Bound the memory an EC shard recovery holds (#11020)

* Reconstruct an EC shard from the shards already on this server

recoverOneRemoteEcShardInterval only ever fanned out to the cached shard
locations, so a server holding shards of the volume still fetched them
over gRPC from itself -- and when the peers were unreachable it could not
reconstruct at all, even holding the whole volume on local disk. Seed the
Reed-Solomon buffers from the locally mounted shards first; each one is a
peer round trip, and an interval-sized buffer, the fan-out no longer needs.

Claude-Session: https://claude.ai/code/session_01SM5ARdPvFcnvGWNpBPgNRN

* Fetch only the EC shards reconstruction still needs

The recovery fan-out read every surviving shard, so a 10+4 volume pulled
13 interval-sized buffers to feed Reed-Solomon 10 -- a third more memory
held, and a third more load asked of peers that were, by definition,
already having trouble. Fetch what is missing, and widen only when some of
those reads fail. A shard reporting the needle deleted ends the walk: the
rest would only answer the same.

Claude-Session: https://claude.ai/code/session_01SM5ARdPvFcnvGWNpBPgNRN

* Bound the bytes EC recovery holds in flight

Recovery is the one read path that multiplies the served bytes: it holds
an interval-sized buffer per shard until Reed-Solomon runs, and a peer that
is slow to fail keeps them all alive for the whole gRPC timeout. Nothing
bounded how many of those fan-outs ran at once, so a transient problem
between volume servers turned every read into a DataShards-fold allocation
and the server died of it -- 64 concurrent 4MB intervals pin 3.6GB, and
that is a small burst.

Charge each recovery against a process-wide budget, so a burst queues on
the semaphore instead of on the heap.

Claude-Session: https://claude.ai/code/session_01SM5ARdPvFcnvGWNpBPgNRN

* Answer a deleted EC needle as deleted, not as a failed recovery

A holder reporting the needle deleted is authoritative: deletes are never
invented and never undone. Recovery already collected that flag, then
dropped it on the branch where too few shards came back -- so a read of a
deleted needle that had to recover surfaced as "cannot recover shard", and
the volume server answered 500 where it owed a 404. Carry the flag out of
the shortfall, and let it decide ahead of the error it came with.

Claude-Session: https://claude.ai/code/session_01SM5ARdPvFcnvGWNpBPgNRN

* Check the encode run of a locally seeded EC shard in Rust

The Rust recovery seeded Reed-Solomon straight from the mounted shards,
without the encode-run check the remote reads and Go's
readLocalEcShardInterval both apply. A volume remounted from a newer
encode between the read's snapshot and its recovery would have fed
mixed-generation bytes into the reconstruction.

Claude-Session: https://claude.ai/code/session_01SM5ARdPvFcnvGWNpBPgNRN

* Say what the recovery budget actually guarantees

Claude-Session: https://claude.ai/code/session_01SM5ARdPvFcnvGWNpBPgNRN

* Seed Rust EC recovery from shards on every local disk

find_ec_volume returns the first disk's EcVolume, so a reconciled volume
whose shards are split across data dirs had the siblings ignored and could
report "cannot recover" while holding enough shards locally. Resolve each
shard together with the disk that owns it, the way Go's recovery already
does, and check that owner's encode run.

Claude-Session: https://claude.ai/code/session_01SM5ARdPvFcnvGWNpBPgNRN

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<Vec<u8>> {
+) -> io::Result<(Vec<u8>, 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<Option<Vec<u8>>> = 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<String>)> = 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!(

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)

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