diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 9b5c0eed3..41f94794b 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/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() diff --git a/weed/operation/lookup.go b/weed/operation/lookup.go index 0580c9f95..eabea475e 100644 --- a/weed/operation/lookup.go +++ b/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) diff --git a/weed/operation/lookup_cache_invalidation_test.go b/weed/operation/lookup_cache_invalidation_test.go new file mode 100644 index 000000000..182951dcd --- /dev/null +++ b/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)) + } +} diff --git a/weed/operation/lookup_vid_cache.go b/weed/operation/lookup_vid_cache.go index e2ff405a9..28730f093 100644 --- a/weed/operation/lookup_vid_cache.go +++ b/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)) +} diff --git a/weed/operation/lookup_vid_cache_test.go b/weed/operation/lookup_vid_cache_test.go index a6ca0a556..8a8b63609 100644 --- a/weed/operation/lookup_vid_cache_test.go +++ b/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 diff --git a/weed/topology/store_replicate.go b/weed/topology/store_replicate.go index 59e8d5c9c..f36868af8 100644 --- a/weed/topology/store_replicate.go +++ b/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) }