Re-check an EC shard map a failed read has disproved (#11023)

* Re-check an EC shard map a failed read has disproved

A read that fails against a cached location drops that shard from the map,
which leaves it one short of complete -- and a map one short is trusted for
seven more minutes. So a moment's trouble between volume servers cost
minutes in which every read of that shard skipped the direct fetch and paid
for a Reed-Solomon recovery instead, at DataShards times the memory and the
peer load.

Mark the map when a read disproves it, and re-check a marked map on the
same eleven-second footing as one that never had enough shards to begin
with. The mark clears on refresh, so it buys one prompt re-check rather
than a master lookup per read. The tiers move into a helper; they were
three overlapping conditions in one expression, and the reading of them
was not obvious.

Rust keeps the entry rather than dropping it -- a dead peer fails fast on
the next attempt, and it was the freshness window, not the entry, hiding a
shard that had moved.

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

* Invalidate the location of an EC shard whose own read failed

Recovery fans out to the other shards, so the one whose direct read just
failed is the only location nothing ever invalidates: a shard that moved to
another server was reconstructed on every read until the map's own window
expired, up to thirty-seven minutes for a map still complete. Mark the map
there too. The entry stays -- a moved shard's old holder fails fast, and
the next refresh is seconds away.

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

* Consume the stale mark before the lookup, not after

A read that fails while the master is answering has disproved the very map
that answer is about to install, and clearing the mark on the refresh's
return swallowed it. Clear it where it is acted on instead. A lookup that
then fails loses the mark, which costs nothing: the refresh time is only
advanced on success, so the next read looks up regardless.

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

* Judge the shard map and consume its mark in one critical section

Reading the mark and clearing it were two separate acquisitions, so a mark
raised between them was cleared by a refresh that had not seen it. In Go
that gap was a few instructions; in Rust the mark was read when the read
first snapshotted the volume and cleared at the decision point, with the
local interval reads in between. Take both under one hold. Rust needs a
mutex rather than an atomic to do it, and no longer carries the mark
through the snapshot.

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

* Put the stale mark back when the lookup does not answer for it

Consuming the mark up front assumed the lookup would supersede it. A
lookup that fails, or comes back with fewer than DataShards holders,
supersedes nothing: the map is unchanged, its refresh time unadvanced, and
with the mark gone the map a read had disproved is trusted for its full
window again on the strength of a lookup that never landed. Put the mark
back on both branches.

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

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<crate::pb::volume_server_pb::EcShardInfo>, Vec<String>) {
     // 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<ShardId, Vec<String>>,
     refreshed_at: Option<Instant>,
+    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<VolumeServerState>, 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<VolumeServerState>,
+    vid: VolumeId,
+    locations: &HashMap<ShardId, Vec<String>>,
+    refreshed_at: Option<Instant>,
+    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<ShardId, Vec<String>> {
+        (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));
+    }
+}

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<Option<std::time::Instant>>,
+    /// 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<bool>,
     /// 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,