diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs index b39368a38..4b1e785bf 100644 --- a/seaweed-volume/src/server/store_ec.rs +++ b/seaweed-volume/src/server/store_ec.rs @@ -121,7 +121,9 @@ pub async fn read_ec_shard_needle_distributed( let mut shard_locations = snapshot.cached_locations.clone(); if any_remote - && needs_refresh( + && claim_shard_locations_refresh( + state, + vid, &shard_locations, snapshot.cache_refreshed_at, snapshot.data_shards as usize, @@ -132,17 +134,21 @@ pub async fn read_ec_shard_needle_distributed( Ok(fresh) => { // A complete reply merges into the cache; an incomplete one // (< data_shards) is left unwritten — keep the prior cache. - if let Some(merged) = - write_back_shard_locations(state, vid, fresh, snapshot.data_shards as usize) + match write_back_shard_locations(state, vid, fresh, snapshot.data_shards as usize) { - shard_locations = merged; + Some(merged) => shard_locations = merged, + // An incomplete reply leaves the cache unwritten and its refresh + // time unadvanced, so the mark this refresh consumed goes back. + None => mark_shard_locations_stale(state, vid), } } Err(e) => { // Lookup failed — proceed with cached values. If cache // is empty, the remote fetch below will fail and we // surface a NotFound (matching Go's behavior when no - // locations are known). + // locations are known). The mark goes back: nothing + // answered for it, and the map stays disproved. + mark_shard_locations_stale(state, vid); tracing::warn!( "ec lookup failed for volume {}: {} — using cached locations ({} entries)", vid.0, @@ -252,7 +258,15 @@ pub async fn scrub_ec_volume_distributed( ) -> (i64, Vec, Vec) { // Phase A — under the Store read lock, run the index scrub and grab the // paths/scalars + shard-location staleness; release the lock before any await. - let (ecx_path, collection, seed_errs, cached_locations, cache_refreshed_at, data_shards, total_shards) = { + let ( + ecx_path, + collection, + seed_errs, + cached_locations, + cache_refreshed_at, + data_shards, + total_shards, + ) = { let store = state.store.read().unwrap(); let ecv = match store.find_ec_volume(vid) { Some(v) => v, @@ -287,10 +301,18 @@ pub async fn scrub_ec_volume_distributed( // cachedLookupEcShardLocations). A partial reply (< data_shards locations, a // master mid-recovery) or a failed lookup is a hard, retryable error — never // overwrite a good cache with a partial map or storm a down master per needle. - if needs_refresh(&cached_locations, cache_refreshed_at, data_shards, total_shards) { + if claim_shard_locations_refresh( + state, + vid, + &cached_locations, + cache_refreshed_at, + data_shards, + total_shards, + ) { match cached_lookup_ec_shard_locations(state, vid).await { Ok(fresh) => { if write_back_shard_locations(state, vid, fresh, data_shards).is_none() { + mark_shard_locations_stale(state, vid); return ( 0, Vec::new(), @@ -302,11 +324,12 @@ pub async fn scrub_ec_volume_distributed( } } Err(e) => { + mark_shard_locations_stale(state, vid); return ( 0, Vec::new(), vec![format!("failed to locate shard via master grpc: {}", e)], - ) + ); } } } @@ -603,6 +626,7 @@ fn build_snapshot( fn needs_refresh( locations: &HashMap>, refreshed_at: Option, + stale: bool, data_shards: usize, total_shards: usize, ) -> bool { @@ -612,16 +636,53 @@ fn needs_refresh( None => return true, }; let shard_count = locations.len(); - if shard_count < data_shards && age < Duration::from_secs(11) { - return false; + // A complete map is trusted longest. One short of data_shards, or one a + // failed read has just disproved, is re-checked promptly: until it is, reads + // keep aiming at a location the shard has left. + let ttl = if stale || shard_count < data_shards { + Duration::from_secs(11) + } else if shard_count == total_shards { + Duration::from_secs(37 * 60) + } else { + Duration::from_secs(7 * 60) + }; + age >= ttl +} + +/// Mark the cached shard map for a prompt re-check after a read failed against +/// one of its locations. Go drops the entry outright in `forgetShardId`, which +/// costs it the direct read until the map is re-learned; here the entry stays +/// (a dead peer just fails fast on the next attempt) and only the freshness +/// window is cut, so a shard that has moved is picked up in seconds either way. +fn mark_shard_locations_stale(state: &Arc, vid: VolumeId) { + let store = state.store.read().unwrap(); + if let Some(ecv) = store.find_ec_volume(vid) { + *ecv.shard_locations_stale.lock().unwrap() = true; } - if shard_count == total_shards && age < Duration::from_secs(37 * 60) { - return false; +} + +/// Decide whether the cached map is due a master lookup and, when it is, consume +/// its stale mark in the same critical section. A mark raised from here on +/// belongs to the next refresh: the read that raised it has disproved the map +/// this lookup is about to install. +fn claim_shard_locations_refresh( + state: &Arc, + vid: VolumeId, + locations: &HashMap>, + refreshed_at: Option, + data_shards: usize, + total_shards: usize, +) -> bool { + let store = state.store.read().unwrap(); + let Some(ecv) = store.find_ec_volume(vid) else { + return needs_refresh(locations, refreshed_at, false, data_shards, total_shards); + }; + let mut stale = ecv.shard_locations_stale.lock().unwrap(); + let refresh = needs_refresh(locations, refreshed_at, *stale, data_shards, total_shards); + if refresh { + *stale = false; } - if shard_count >= data_shards && age < Duration::from_secs(7 * 60) { - return false; - } - true + refresh } async fn cached_lookup_ec_shard_locations( @@ -757,6 +818,9 @@ async fn fetch_one_interval( sources, e ); + // Reconstruction below skips this very shard, so nothing else + // invalidates the location that just failed. + mark_shard_locations_stale(state, vid); } } } @@ -1058,6 +1122,7 @@ async fn recover_one_remote_ec_shard_interval( needle_id, e ); + mark_shard_locations_stale(state, vid); } } } @@ -1332,3 +1397,34 @@ async fn drain_copy_stream( } Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + fn locations(count: usize) -> HashMap> { + (0..count) + .map(|sid| (sid as ShardId, vec!["127.0.0.1:8080".to_string()])) + .collect() + } + + #[test] + fn needs_refresh_re_checks_a_map_a_failed_read_disproved() { + let just_now = Some(Instant::now()); + let aged = Some(Instant::now() - Duration::from_secs(12)); + + // A complete map is trusted for a long time, and one shard short still + // outlasts a 12-second gap. + assert!(!needs_refresh(&locations(14), aged, false, 10, 14)); + assert!(!needs_refresh(&locations(13), aged, false, 10, 14)); + // Disproved by a read, the same maps are re-checked within seconds. + assert!(needs_refresh(&locations(14), aged, true, 10, 14)); + assert!(needs_refresh(&locations(13), aged, true, 10, 14)); + // But the mark buys one prompt re-check, not a lookup per read. + assert!(!needs_refresh(&locations(14), just_now, true, 10, 14)); + // A map short of the data shards is re-checked promptly regardless. + assert!(needs_refresh(&locations(9), aged, false, 10, 14)); + // An unrefreshed cache always looks up. + assert!(needs_refresh(&locations(0), None, false, 10, 14)); + } +} diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index 3f1717a79..709d8f485 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs @@ -55,6 +55,13 @@ pub struct EcVolume { /// `cached_lookup_ec_shard_locations` (mirrors Go's /// `ShardLocationsRefreshTime`). pub shard_locations_refresh_time: std::sync::Mutex>, + /// Marks the map for a prompt re-check: a read that failed against a cached + /// location has disproved what the map claims, and the normal freshness + /// window is far too long to serve from a map known to be wrong. Mirrors the + /// invalidation Go's `forgetShardId` performs. A mutex rather than an atomic + /// so the refresh can judge the map and consume the mark in one critical + /// section, and a mark raised meanwhile survives for the next refresh. + pub shard_locations_stale: std::sync::Mutex, /// EC volume expiration time (unix epoch seconds), set during EC encode from TTL. pub expire_at_sec: u64, /// Encode-run identity (unix nanos) loaded from the .vif EcShardConfig. A read @@ -196,6 +203,7 @@ impl EcVolume { ecx_actual_dir: dir_idx.to_string(), shard_locations: std::sync::RwLock::new(HashMap::new()), shard_locations_refresh_time: std::sync::Mutex::new(None), + shard_locations_stale: std::sync::Mutex::new(false), expire_at_sec, encode_ts_ns, bitrot: None, diff --git a/weed/storage/erasure_coding/ec_volume.go b/weed/storage/erasure_coding/ec_volume.go index 2884afb8f..8bb4a4e8f 100644 --- a/weed/storage/erasure_coding/ec_volume.go +++ b/weed/storage/erasure_coding/ec_volume.go @@ -35,14 +35,18 @@ type EcVolume struct { Shards []*EcVolumeShard ShardLocations map[ShardId][]pb.ServerAddress ShardLocationsRefreshTime time.Time - ShardLocationsLock sync.RWMutex - Version needle.Version - ecjFile *os.File - ecjFileAccessLock sync.Mutex - diskType types.DiskType - datFileSize int64 - ExpireAtSec uint64 //ec volume destroy time, calculated from the ec volume was created - ECContext *ECContext // EC encoding parameters + // ShardLocationsStale marks the map for a prompt re-check: a read that failed + // against a cached location has disproved what the map claims, and the normal + // freshness window is far too long to serve from a map known to be wrong. + ShardLocationsStale bool + ShardLocationsLock sync.RWMutex + Version needle.Version + ecjFile *os.File + ecjFileAccessLock sync.Mutex + diskType types.DiskType + datFileSize int64 + 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. diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 5ef495bba..dc1cbc49a 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -532,6 +532,10 @@ func (s *Store) readOneEcShardInterval(needleId types.NeedleId, ecVolume *erasur return } glog.V(0).Infof("read remote ec shard %d.%d locations: %v", ecVolume.VolumeId, shardId, err) + // Recovery below skips this very shard, so nothing else invalidates the + // location that just failed -- and a shard that has moved would otherwise + // be reconstructed on every read until the map's own window expires. + markShardLocationsStale(ecVolume) } // try reading by recovering from other shards @@ -548,9 +552,34 @@ func forgetShardId(ecVolume *erasure_coding.EcVolume, shardId erasure_coding.Sha // failed to access the source data nodes, clear it up ecVolume.ShardLocationsLock.Lock() delete(ecVolume.ShardLocations, shardId) + ecVolume.ShardLocationsStale = true ecVolume.ShardLocationsLock.Unlock() } +// markShardLocationsStale flags the cached map for a prompt re-check after a +// read failed against one of its locations. Unlike forgetShardId it keeps the +// entry: a direct read is worth retrying, since a dead peer fails fast. +func markShardLocationsStale(ecVolume *erasure_coding.EcVolume) { + ecVolume.ShardLocationsLock.Lock() + ecVolume.ShardLocationsStale = true + ecVolume.ShardLocationsLock.Unlock() +} + +// ecShardLocationsTTL is how long a cached shard map is trusted. A complete map +// is trusted longest. One short of DataShards, or one a failed read has just +// invalidated, is re-checked promptly: until it is, every read of the dropped +// shard skips the direct fetch and pays for a Reed-Solomon recovery instead. +func ecShardLocationsTTL(shardCount int, stale bool, ecCtx *erasure_coding.ECContext) time.Duration { + switch { + case stale || shardCount < ecCtx.DataShards: + return 11 * time.Second + case shardCount == ecCtx.Total(): + return 37 * time.Minute + default: + return 7 * time.Minute + } +} + 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 @@ -561,20 +590,20 @@ func (s *Store) cachedLookupEcShardLocations(ecVolume *erasure_coding.EcVolume) 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) - 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 + // Judge the map and consume its mark in one critical section, so a mark raised + // from here on belongs to the next refresh rather than being cleared by this + // one -- the read that raised it has disproved the map this lookup is about to + // install. Recover goroutines mutate all three fields via forgetShardId, so an + // unguarded read would race a concurrent map write besides. + ecVolume.ShardLocationsLock.Lock() + stale := ecVolume.ShardLocationsStale + ttl := ecShardLocationsTTL(len(ecVolume.ShardLocations), stale, ecCtx) + fresh := ecVolume.ShardLocationsRefreshTime.Add(ttl).After(time.Now()) + if !fresh { + ecVolume.ShardLocationsStale = false + } + ecVolume.ShardLocationsLock.Unlock() + if fresh { return nil } @@ -605,6 +634,14 @@ func (s *Store) cachedLookupEcShardLocations(ecVolume *erasure_coding.EcVolume) return nil }) + if err != nil { + // The lookup this mark was consumed for did not answer, and the refresh time + // is only advanced on success -- so without putting the mark back, a map a + // read had already disproved would be trusted for its full window again. + // Marking unconditionally is safe: where no mark was consumed the refresh + // time is stale anyway, and the next call looks up whatever the mark says. + markShardLocationsStale(ecVolume) + } return } diff --git a/weed/storage/store_ec_location_cache_test.go b/weed/storage/store_ec_location_cache_test.go new file mode 100644 index 000000000..c2bb1b5c5 --- /dev/null +++ b/weed/storage/store_ec_location_cache_test.go @@ -0,0 +1,330 @@ +package storage + +import ( + "bytes" + "context" + "net" + "os" + "sync/atomic" + "testing" + "time" + + "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/master_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" +) + +// countingMaster answers LookupEcVolume with one holder per shard and counts +// how often it was asked. +type countingMaster struct { + master_pb.UnimplementedSeaweedServer + lookups atomic.Int64 + // onLookup runs inside the RPC, standing in for whatever else the volume + // server is doing while it waits on the master. + onLookup func() + // failLookup answers with an error instead of a shard map. + failLookup atomic.Bool +} + +func (m *countingMaster) LookupEcVolume(_ context.Context, req *master_pb.LookupEcVolumeRequest) (*master_pb.LookupEcVolumeResponse, error) { + m.lookups.Add(1) + if m.onLookup != nil { + m.onLookup() + } + if m.failLookup.Load() { + return nil, status.Error(codes.Unavailable, "master is having a moment") + } + resp := &master_pb.LookupEcVolumeResponse{VolumeId: req.VolumeId} + for shardId := 0; shardId < erasure_coding.TotalShardsCount; shardId++ { + resp.ShardIdLocations = append(resp.ShardIdLocations, &master_pb.LookupEcVolumeResponse_EcShardIdLocation{ + ShardId: uint32(shardId), + Locations: []*master_pb.Location{{Url: "127.0.0.1:1", GrpcPort: 2}}, + }) + } + return resp, nil +} + +func startCountingMaster(t *testing.T) (*countingMaster, pb.ServerAddress) { + t.Helper() + master := &countingMaster{} + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + srv := grpc.NewServer() + master_pb.RegisterSeaweedServer(srv, master) + go srv.Serve(lis) + t.Cleanup(srv.Stop) + return master, pb.NewServerAddressWithGrpcPort("127.0.0.1:1", lis.Addr().(*net.TCPAddr).Port) +} + +// rewindShardLocationsRefresh ages the cache by d so a test can reach a TTL +// without waiting one out. +func rewindShardLocationsRefresh(ecVolume *erasure_coding.EcVolume, d time.Duration) { + ecVolume.ShardLocationsLock.Lock() + ecVolume.ShardLocationsRefreshTime = ecVolume.ShardLocationsRefreshTime.Add(-d) + ecVolume.ShardLocationsLock.Unlock() +} + +// A read that fails against a cached location leaves the map claiming something +// it has just disproved, so the next lookup goes back to the master within +// seconds rather than serving the hole for the rest of a 7-minute window. +func TestCachedLookupEcShardLocationsRefreshesAfterAFailedRead(t *testing.T) { + master, masterAddr := startCountingMaster(t) + store := &Store{ + MasterAddress: masterAddr, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + ecVolume := &erasure_coding.EcVolume{ + VolumeId: 7, + ShardLocations: make(map[erasure_coding.ShardId][]pb.ServerAddress), + } + + if err := store.cachedLookupEcShardLocations(ecVolume); err != nil { + t.Fatalf("first lookup: %v", err) + } + if got := master.lookups.Load(); got != 1 { + t.Fatalf("cold cache asked the master %d times, want 1", got) + } + + // A map nothing has disproved is trusted for minutes, whether or not it is + // complete -- dropping a shard by hand is not the same as a read failing. + rewindShardLocationsRefresh(ecVolume, 12*time.Second) + ecVolume.ShardLocationsLock.Lock() + delete(ecVolume.ShardLocations, erasure_coding.ShardId(2)) + ecVolume.ShardLocationsLock.Unlock() + if err := store.cachedLookupEcShardLocations(ecVolume); err != nil { + t.Fatalf("second lookup: %v", err) + } + if got := master.lookups.Load(); got != 1 { + t.Errorf("a 12s-old map asked the master %d times, want it left alone", got) + } + + forgetShardId(ecVolume, erasure_coding.ShardId(3)) + if err := store.cachedLookupEcShardLocations(ecVolume); err != nil { + t.Fatalf("lookup after a failed read: %v", err) + } + if got := master.lookups.Load(); got != 2 { + t.Errorf("a map disproved by a failed read asked the master %d times, want 2", got) + } + + ecVolume.ShardLocationsLock.RLock() + _, restored := ecVolume.ShardLocations[erasure_coding.ShardId(3)] + ecVolume.ShardLocationsLock.RUnlock() + if !restored { + t.Error("shard 3 is still missing from the map after the refresh") + } +} + +// A failed read buys one prompt re-check, not a lookup on every read that +// follows: the refresh clears the mark and the map is trusted again. +func TestCachedLookupEcShardLocationsSettlesAfterRefresh(t *testing.T) { + master, masterAddr := startCountingMaster(t) + store := &Store{ + MasterAddress: masterAddr, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + ecVolume := &erasure_coding.EcVolume{ + VolumeId: 8, + ShardLocations: make(map[erasure_coding.ShardId][]pb.ServerAddress), + } + if err := store.cachedLookupEcShardLocations(ecVolume); err != nil { + t.Fatalf("first lookup: %v", err) + } + rewindShardLocationsRefresh(ecVolume, 12*time.Second) + forgetShardId(ecVolume, erasure_coding.ShardId(3)) + if err := store.cachedLookupEcShardLocations(ecVolume); err != nil { + t.Fatalf("lookup after a failed read: %v", err) + } + for i := 0; i < 5; i++ { + if err := store.cachedLookupEcShardLocations(ecVolume); err != nil { + t.Fatalf("settled lookup: %v", err) + } + } + if got := master.lookups.Load(); got != 2 { + t.Errorf("asked the master %d times, want 2: one cold, one after the failed read", got) + } +} + +func TestEcShardLocationsTTL(t *testing.T) { + ecCtx := erasure_coding.NewDefaultECContext("", 1) + for _, tc := range []struct { + name string + shardCount int + stale bool + want time.Duration + }{ + {"complete", ecCtx.Total(), false, 37 * time.Minute}, + {"one shard short", ecCtx.Total() - 1, false, 7 * time.Minute}, + {"below the data shards", ecCtx.DataShards - 1, false, 11 * time.Second}, + {"complete but disproved by a read", ecCtx.Total(), true, 11 * time.Second}, + {"one shard short and disproved", ecCtx.Total() - 1, true, 11 * time.Second}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := ecShardLocationsTTL(tc.shardCount, tc.stale, ecCtx); got != tc.want { + t.Errorf("ttl = %v, want %v", got, tc.want) + } + }) + } +} + +// oneBadShardPeer serves every shard out of an EC volume's files except one, +// standing in for a shard that has moved off the address the map still names. +type oneBadShardPeer struct { + volume_server_pb.UnimplementedVolumeServerServer + baseFileName string + badShard erasure_coding.ShardId +} + +func (p *oneBadShardPeer) VolumeEcShardRead(req *volume_server_pb.VolumeEcShardReadRequest, stream volume_server_pb.VolumeServer_VolumeEcShardReadServer) error { + if erasure_coding.ShardId(req.ShardId) == p.badShard { + return status.Errorf(codes.NotFound, "shard %d is not here", req.ShardId) + } + 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}) +} + +// Recovery skips the shard whose direct read failed, so that failure is the one +// thing that never invalidates the map. The read still succeeds off the other +// shards, and the location it just disproved has to be re-checked anyway. +func TestReadOneEcShardIntervalMarksTheMapAfterADirectReadFails(t *testing.T) { + const badShard = erasure_coding.ShardId(3) + baseFileName, n := writeEcVolumeFiles(t, t.TempDir(), 7) + + 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, &oneBadShardPeer{baseFileName: baseFileName, badShard: badShard}) + go srv.Serve(lis) + t.Cleanup(srv.Stop) + + store := &Store{grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials())} + ecVolume := &erasure_coding.EcVolume{ + VolumeId: 7, + ShardLocations: make(map[erasure_coding.ShardId][]pb.ServerAddress), + } + seedShardLocations(ecVolume, pb.NewServerAddressWithGrpcPort("127.0.0.1:1", lis.Addr().(*net.TCPAddr).Port)) + + interval := erasure_coding.Interval{BlockIndex: int(badShard), Size: 4096, IsLargeBlock: true} + got := make([]byte, interval.Size) + if _, err := store.readOneEcShardInterval(n.Id, ecVolume, interval, got); err != nil { + t.Fatalf("read interval: %v", err) + } + + ecVolume.ShardLocationsLock.RLock() + stale := ecVolume.ShardLocationsStale + _, stillMapped := ecVolume.ShardLocations[badShard] + ecVolume.ShardLocationsLock.RUnlock() + if !stale { + t.Error("a failed direct read left the map trusted, so the moved shard stays hidden until the window expires") + } + if !stillMapped { + t.Error("shard 3 was dropped from the map; the direct read is worth retrying") + } + + want := make([]byte, len(got)) + f, err := os.Open(baseFileName + erasure_coding.ToExt(int(badShard))) + if err != nil { + t.Fatalf("open shard %d: %v", badShard, err) + } + defer f.Close() + if _, err := f.ReadAt(want, 0); err != nil { + t.Fatalf("read shard %d: %v", badShard, err) + } + if !bytes.Equal(got, want) { + t.Error("the interval recovered from the other shards does not match the shard on disk") + } +} + +// A read that fails while the master is answering has disproved the very map +// that answer is about to install, so the refresh must not clear its mark. +func TestCachedLookupEcShardLocationsKeepsAMarkRaisedDuringTheLookup(t *testing.T) { + master, masterAddr := startCountingMaster(t) + store := &Store{ + MasterAddress: masterAddr, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + ecVolume := &erasure_coding.EcVolume{ + VolumeId: 9, + ShardLocations: make(map[erasure_coding.ShardId][]pb.ServerAddress), + } + + master.onLookup = func() { markShardLocationsStale(ecVolume) } + if err := store.cachedLookupEcShardLocations(ecVolume); err != nil { + t.Fatalf("first lookup: %v", err) + } + master.onLookup = nil + + ecVolume.ShardLocationsLock.RLock() + stale := ecVolume.ShardLocationsStale + ecVolume.ShardLocationsLock.RUnlock() + if !stale { + t.Fatal("the refresh swallowed a mark raised while it was in flight") + } + + // And the mark shortens the window, so the next read re-checks in seconds + // rather than trusting a map already disproved. + rewindShardLocationsRefresh(ecVolume, 12*time.Second) + if err := store.cachedLookupEcShardLocations(ecVolume); err != nil { + t.Fatalf("lookup after the swallowed mark: %v", err) + } + if got := master.lookups.Load(); got != 2 { + t.Errorf("asked the master %d times, want 2", got) + } +} + +// A refresh that never answered cannot be treated as one that did: the mark it +// consumed goes back, or the map it was meant to correct is trusted for its +// whole window on the strength of a lookup that failed. +func TestCachedLookupEcShardLocationsKeepsTheMarkWhenTheLookupFails(t *testing.T) { + master, masterAddr := startCountingMaster(t) + store := &Store{ + MasterAddress: masterAddr, + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + ecVolume := &erasure_coding.EcVolume{ + VolumeId: 10, + ShardLocations: make(map[erasure_coding.ShardId][]pb.ServerAddress), + } + if err := store.cachedLookupEcShardLocations(ecVolume); err != nil { + t.Fatalf("first lookup: %v", err) + } + + rewindShardLocationsRefresh(ecVolume, 12*time.Second) + markShardLocationsStale(ecVolume) + master.failLookup.Store(true) + if err := store.cachedLookupEcShardLocations(ecVolume); err == nil { + t.Fatal("expected the lookup to fail") + } + + ecVolume.ShardLocationsLock.RLock() + stale := ecVolume.ShardLocationsStale + ecVolume.ShardLocationsLock.RUnlock() + if !stale { + t.Error("the failed lookup kept the mark it consumed") + } + + master.failLookup.Store(false) + if err := store.cachedLookupEcShardLocations(ecVolume); err != nil { + t.Fatalf("retry after the failed lookup: %v", err) + } + if got := master.lookups.Load(); got != 3 { + t.Errorf("asked the master %d times, want 3: a failed lookup must not leave the disproved map trusted", got) + } +}