volume: drop stale volume-location cache on under-replication (#10185)

* volume: drop stale volume-location cache on under-replication

A replicated write looks up the volume's locations and caches them for 10
minutes. When the master briefly reports fewer replicas than the copy count
(e.g. a stale heartbeat drops a just-added volume), that under-replicated
result got cached, so every write failed with "replicating operations is less
than replication copy count" until the entry expired -- long after the master
re-registered the replica.

Invalidate the cached entry when the location count is below the copy count, so
the next write re-queries the master and recovers as soon as it heals.

* volume: mirror the replication copy-count guard in seaweed-volume

do_replicated_request accepted a write even when the master reported fewer
locations than the volume's copy count, silently under-replicating. Reject it,
matching Go's GetWritableRemoteReplications. lookup_volume is uncached, so the
next write recovers as soon as the missing replica re-registers.

seaweed-volume/src/server/handlers.rs

@@ -546,6 +546,23 @@ async fn do_replicated_request(
     .await
     .map_err(|e| format!("lookup volume failed: {}", e))?;
 
+    // Mirror Go's GetWritableRemoteReplications: reject when the master reports fewer replicas than
+    // the copy count. lookup_volume is uncached, so recovery is immediate once the replica re-registers.
+    let copy_count = {
+        let store = state.store.read().unwrap();
+        store.find_volume(VolumeId(vid)).map_or(1, |(_, v)| {
+            v.super_block.replica_placement.get_copy_count()
+        })
+    };
+    if locations.len() < copy_count as usize {
+        return Err(format!(
+            "replicating operations [{}] is less than volume {} replication copy count [{}]",
+            locations.len(),
+            vid,
+            copy_count
+        ));
+    }
+
     let self_http = to_http_address(&state.self_url);
     let remote_locations: Vec<_> = locations
         .into_iter()

weed/operation/lookup.go

@@ -61,6 +61,11 @@ func LookupVolumeId(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vid st
 	return results[vid], err
 }
 
+// InvalidateVolumeIdLocationCache drops cached locations for vid so the next lookup re-queries the master.
+func InvalidateVolumeIdLocationCache(vid string) {
+	vc.Delete(vid)
+}
+
 // LookupVolumeIds find volume locations by cache and actual lookup
 func LookupVolumeIds(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids []string) (map[string]*LookupResult, error) {
 	ret := make(map[string]*LookupResult)

weed/operation/lookup_cache_invalidation_test.go

@@ -0,0 +1,83 @@
+package operation
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+type fakeLookupServer struct {
+	master_pb.UnimplementedSeaweedServer
+	mu        sync.Mutex
+	calls     int
+	locations int
+}
+
+func (s *fakeLookupServer) LookupVolume(_ context.Context, req *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.calls++
+	resp := &master_pb.LookupVolumeResponse{}
+	for _, vid := range req.VolumeOrFileIds {
+		var locs []*master_pb.Location
+		for i := 0; i < s.locations; i++ {
+			locs = append(locs, &master_pb.Location{Url: fmt.Sprintf("10.0.0.%d:8080", i)})
+		}
+		resp.VolumeIdLocations = append(resp.VolumeIdLocations, &master_pb.LookupVolumeResponse_VolumeIdLocation{
+			VolumeOrFileId: vid,
+			Locations:      locs,
+		})
+	}
+	return resp, nil
+}
+
+// TestLookupVolumeIdCacheInvalidation reproduces the stale-lookup scenario: a
+// volume that briefly reports too few replicas would be cached for the full TTL,
+// so writes kept failing even after the master re-registered the missing replica.
+// InvalidateVolumeIdLocationCache forces the next lookup to re-query the master.
+func TestLookupVolumeIdCacheInvalidation(t *testing.T) {
+	fake := &fakeLookupServer{locations: 1}
+	master := startFakeMasterServer(t, fake)
+	masterFn := func(context.Context) pb.ServerAddress { return master }
+	dial := grpc.WithTransportCredentials(insecure.NewCredentials())
+
+	const vid = "778899"
+	InvalidateVolumeIdLocationCache(vid)
+
+	// first lookup queries the master and caches the single, under-replicated location
+	r1, err := LookupVolumeId(masterFn, dial, vid)
+	if err != nil {
+		t.Fatalf("first lookup: %v", err)
+	}
+	if len(r1.Locations) != 1 {
+		t.Fatalf("first lookup locations = %d, want 1", len(r1.Locations))
+	}
+
+	// a second lookup is served from cache, so the master is not queried again
+	if _, err := LookupVolumeId(masterFn, dial, vid); err != nil {
+		t.Fatalf("second lookup: %v", err)
+	}
+	fake.mu.Lock()
+	calls := fake.calls
+	fake.locations = 2 // master heals: both replicas are registered again
+	fake.mu.Unlock()
+	if calls != 1 {
+		t.Fatalf("expected a cache hit, master queried %d times", calls)
+	}
+
+	// invalidation drops the stale entry, so the next lookup re-queries and sees both replicas
+	InvalidateVolumeIdLocationCache(vid)
+	r3, err := LookupVolumeId(masterFn, dial, vid)
+	if err != nil {
+		t.Fatalf("third lookup: %v", err)
+	}
+	if len(r3.Locations) != 2 {
+		t.Fatalf("after invalidation locations = %d, want 2", len(r3.Locations))
+	}
+}

weed/operation/lookup_vid_cache.go

@@ -60,3 +60,13 @@ func (vc *VidCache) Set(vid string, locations []Location, duration time.Duration
 		NextRefreshTime: time.Now().Add(duration),
 	}
 }
+
+func (vc *VidCache) Delete(vid string) {
+	id, err := strconv.ParseUint(vid, 10, 32)
+	if err != nil || id == 0 {
+		return
+	}
+	vc.Lock()
+	defer vc.Unlock()
+	delete(vc.cache, uint32(id))
+}

weed/operation/lookup_vid_cache_test.go

@@ -25,6 +25,23 @@ func TestCaching(t *testing.T) {
 	}
 }
 
+// a stale under-replicated result must not linger for the full cache TTL
+func TestCachingDelete(t *testing.T) {
+	var vc VidCache
+	locations := []Location{{Url: "a.com:8080"}}
+	vc.Set("123", locations, time.Minute)
+	if ret, _ := vc.Get("123"); ret == nil {
+		t.Fatal("expected vid 123 to be cached")
+	}
+	vc.Delete("123")
+	if ret, _ := vc.Get("123"); ret != nil {
+		t.Fatal("expected vid 123 to be evicted after Delete")
+	}
+	// deleting a missing or out-of-range id must not panic
+	vc.Delete("123")
+	vc.Delete("4294967296")
+}
+
 // a single large volume id must not allocate an entry per id below it
 func TestCachingLargeVolumeId(t *testing.T) {
 	var vc VidCache

weed/topology/store_replicate.go

@@ -292,6 +292,8 @@ func GetWritableRemoteReplications(s *storage.Store, grpcDialOption grpc.DialOpt
 		// has one local and has remote replications
 		copyCount := v.ReplicaPlacement.GetCopyCount()
 		if len(lookupResult.Locations) < copyCount {
+			// drop the stale cache so the next write re-queries the master once it re-registers the missing replica
+			operation.InvalidateVolumeIdLocationCache(volumeId.String())
 			err = fmt.Errorf("replicating operations [%d] is less than volume %d replication copy count [%d]",
 				len(lookupResult.Locations), volumeId, copyCount)
 		}