diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index 36ccad3768..8bd16cb1ec 100644 --- a/weed/s3api/s3api_server.go +++ b/weed/s3api/s3api_server.go @@ -248,6 +248,8 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl DiscoveryInterval: 5 * time.Minute, }) masterClient.SetOnPeerUpdateFn(filerClient.OnPeerUpdate) + masterClient.SetOnVolumeLocationChangeFn(filerClient.OnVolumeLocationChange) + masterClient.SetOnVidMapResetFn(filerClient.OnVidMapReset) // Start the master client connection loop - required for GetMaster() to work go masterClient.KeepConnectedToMaster(context.Background()) diff --git a/weed/wdclient/filer_client.go b/weed/wdclient/filer_client.go index 55b3204f8f..cf32328f0c 100644 --- a/weed/wdclient/filer_client.go +++ b/weed/wdclient/filer_client.go @@ -527,6 +527,25 @@ func (fc *FilerClient) applyDiscoveredFilersLocked(discoveredFilers map[pb.Serve } } +// OnVolumeLocationChange applies master-pushed changes to the lookup-fed cache, +// so a retired server stops being tried before a read against it fails. Adds +// only lift the removal fence; locations are still learned by lookup. +func (fc *FilerClient) OnVolumeLocationChange(loc Location, addedVids, deletedVids []uint32) { + for _, vid := range deletedVids { + fc.tombstone(vid, loc) + fc.deleteLocation(vid, loc) + } + for _, vid := range addedVids { + fc.untombstone(vid, loc) + } +} + +// OnVidMapReset drops the cache, since removals made while the master link +// was down are never pushed. +func (fc *FilerClient) OnVidMapReset() { + fc.clearVidMap() +} + // GetLookupFileIdFunction returns a lookup function with URL preference handling func (fc *FilerClient) GetLookupFileIdFunction() LookupFileIdFunctionType { if fc.urlPreference == PreferUrl { diff --git a/weed/wdclient/filer_client_test.go b/weed/wdclient/filer_client_test.go index 1bfa56223f..97e23dd9cb 100644 --- a/weed/wdclient/filer_client_test.go +++ b/weed/wdclient/filer_client_test.go @@ -25,6 +25,7 @@ func newTestFilerClient(addrs ...pb.ServerAddress) *FilerClient { return &FilerClient{ filerAddresses: addrs, filerHealth: health, + vidMapClient: newVidMapClient(nil, "", 0), } } @@ -389,3 +390,174 @@ func TestLookupVolumeIdsCountsOnlyFilerFailures(t *testing.T) { }) } } + +func TestOnVolumeLocationChangeDropsRetiredServer(t *testing.T) { + fc := newTestFilerClient() + retired := Location{Url: "retired:8080"} + live := Location{Url: "live:8080"} + fc.addLocation(7, retired) + fc.addLocation(7, live) + fc.addLocation(8, retired) + + fc.OnVolumeLocationChange(retired, nil, []uint32{7, 8, 9}) + + locations, found := fc.GetLocations(7) + if !found || len(locations) != 1 || locations[0].Url != live.Url { + t.Errorf("vid 7 kept only the surviving location: found=%v %v", found, locations) + } + if locations, found := fc.GetLocations(8); found && len(locations) > 0 { + t.Errorf("vid 8 lost its last location but stayed cached: %v", locations) + } +} + +func TestOnVidMapResetClearsCachedLocations(t *testing.T) { + fc := newTestFilerClient() + fc.addLocation(7, Location{Url: "live:8080"}) + + fc.OnVidMapReset() + + if locations, found := fc.GetLocations(7); found && len(locations) > 0 { + t.Errorf("vid 7 survived the reset: %v", locations) + } +} + +func TestOnVolumeLocationChangeReaddClearsTombstone(t *testing.T) { + retired := Location{Url: "retired:8080"} + fc := newTestFilerClient() + fc.vidMapClient = newVidMapClient(&testLocationProvider{ + locations: map[string][]Location{"5": {retired}}, + }, "", 0) + ctx := context.Background() + + fc.OnVolumeLocationChange(retired, nil, []uint32{5}) + if locations, _ := fc.LookupVolumeIdsWithFallback(ctx, []string{"5"}); len(locations["5"]) > 0 { + t.Fatalf("removed location still served: %v", locations) + } + + fc.OnVolumeLocationChange(retired, []uint32{5}, nil) + locations, err := fc.LookupVolumeIdsWithFallback(ctx, []string{"5"}) + if err != nil || len(locations["5"]) != 1 || locations["5"][0].Url != retired.Url { + t.Fatalf("re-added location not served again: %v, err %v", locations, err) + } +} + +func hookedFilerClient(mc *MasterClient, provider VolumeLocationProvider) *FilerClient { + fc := newTestFilerClient() + fc.vidMapClient = newVidMapClient(provider, "", 0) + mc.SetOnVolumeLocationChangeFn(fc.OnVolumeLocationChange) + mc.SetOnVidMapResetFn(fc.OnVidMapReset) + return fc +} + +func expectServed(t *testing.T, fc *FilerClient, vid string, want ...Location) { + t.Helper() + locations, err := fc.LookupVolumeIdsWithFallback(context.Background(), []string{vid}) + got := locations[vid] + if len(want) == 0 { + if len(got) != 0 { + t.Fatalf("volume %s still served from %v", vid, got) + } + return + } + if err != nil || len(got) != len(want) { + t.Fatalf("volume %s served from %v, err %v, want %v", vid, got, err, want) + } + for i := range want { + if got[i].Url != want[i].Url { + t.Fatalf("volume %s served from %v, want %v", vid, got, want) + } + } +} + +// ec.decode mounts the regular copy on a shard server before the shards come +// off it, so the later shard removal must leave that copy servable. +func TestDecodedCopyOutlivesShardRemoval(t *testing.T) { + a := Location{Url: "a:8080"} + b := Location{Url: "b:8080"} + provider := &testLocationProvider{locations: map[string][]Location{"5": {a, b}}} + mc := moveClient() + fc := hookedFilerClient(mc, provider) + pushVolumeLocation(mc, a.Url, nil, nil, []uint32{5}, nil) + pushVolumeLocation(mc, b.Url, nil, nil, []uint32{5}, nil) + expectServed(t, fc, "5", a, b) + + pushVolumeLocation(mc, a.Url, []uint32{5}, nil, nil, nil) + pushVolumeLocation(mc, b.Url, nil, nil, nil, []uint32{5}) + pushVolumeLocation(mc, a.Url, nil, nil, nil, []uint32{5}) + + provider.locations["5"] = []Location{a, b} + expectServed(t, fc, "5", a) +} + +// ec.encode registers the shards before the regular copy is deleted from the +// same server, so the server stays a read candidate for the ec volume. +func TestShardServerOutlivesRegularRemoval(t *testing.T) { + a := Location{Url: "a:8080"} + provider := &testLocationProvider{locations: map[string][]Location{"5": {a}}} + mc := moveClient() + fc := hookedFilerClient(mc, provider) + pushVolumeLocation(mc, a.Url, []uint32{5}, nil, nil, nil) + expectServed(t, fc, "5", a) + + pushVolumeLocation(mc, a.Url, nil, nil, []uint32{5}, nil) + pushVolumeLocation(mc, a.Url, nil, []uint32{5}, nil, nil) + + expectServed(t, fc, "5", a) +} + +// The filer caches shard servers like regular copies, so the last shard +// leaving a server retires it from the cache too. +func TestRetiredShardServerLeavesCache(t *testing.T) { + a := Location{Url: "a:8080"} + b := Location{Url: "b:8080"} + provider := &testLocationProvider{locations: map[string][]Location{"5": {a, b}}} + mc := moveClient() + fc := hookedFilerClient(mc, provider) + pushVolumeLocation(mc, a.Url, nil, nil, []uint32{5}, nil) + pushVolumeLocation(mc, b.Url, nil, nil, []uint32{5}, nil) + expectServed(t, fc, "5", a, b) + + pushVolumeLocation(mc, a.Url, nil, nil, nil, []uint32{5}) + + if cached, _ := fc.GetLocations(5); len(cached) != 1 || cached[0].Url != b.Url { + t.Fatalf("cache holds %v, want only %v", cached, b) + } + fc.InvalidateCache("5,01") + expectServed(t, fc, "5", b) +} + +// A volume moved to another server while the old one stays up: the old +// server would proxy the read and succeed, so no failed read would ever +// evict it. The pushed removal has to. +func TestMovedVolumeLeavesOldServer(t *testing.T) { + old := Location{Url: "old:8080"} + moved := Location{Url: "new:8080"} + provider := &testLocationProvider{locations: map[string][]Location{"5": {old}}} + mc := moveClient() + fc := hookedFilerClient(mc, provider) + pushVolumeLocation(mc, old.Url, []uint32{5}, nil, nil, nil) + expectServed(t, fc, "5", old) + + pushVolumeLocation(mc, moved.Url, []uint32{5}, nil, nil, nil) + pushVolumeLocation(mc, old.Url, nil, []uint32{5}, nil, nil) + + provider.locations["5"] = []Location{old, moved} + expectServed(t, fc, "5", moved) +} + +// A server back at the same URL after a master reconnect is re-told as an +// add, which lifts the fence its earlier removal left. +func TestReconnectReplayLiftsFence(t *testing.T) { + a := Location{Url: "a:8080"} + provider := &testLocationProvider{locations: map[string][]Location{"5": {a}}} + mc := moveClient() + fc := hookedFilerClient(mc, provider) + pushVolumeLocation(mc, a.Url, []uint32{5}, nil, nil, nil) + pushVolumeLocation(mc, a.Url, nil, []uint32{5}, nil, nil) + expectServed(t, fc, "5") + + mc.resetVidMap() + pushVolumeLocation(mc, a.Url, []uint32{5}, nil, nil, nil) + + expectServed(t, fc, "5", a) +} diff --git a/weed/wdclient/masterclient.go b/weed/wdclient/masterclient.go index 91421d203c..e738c38871 100644 --- a/weed/wdclient/masterclient.go +++ b/weed/wdclient/masterclient.go @@ -176,6 +176,14 @@ type MasterClient struct { OnLockRingUpdateLock sync.RWMutex OnMasterChange func(previous, current pb.ServerAddress) OnMasterChangeLock sync.RWMutex + // OnVolumeLocationChange hears, per pushed message, the vids loc added and + // the vids loc no longer serves in any form. + OnVolumeLocationChange func(loc Location, addedVids, deletedVids []uint32) + OnVolumeLocationChangeLock sync.RWMutex + // OnVidMapReset fires when a (re)connect makes the master re-tell its view; + // removals made while disconnected are never pushed. + OnVidMapReset func() + OnVidMapResetLock sync.RWMutex // streamLock publishes the stream only after its registration send, so a // leave sent afterwards never races that send. @@ -271,6 +279,29 @@ func (mc *MasterClient) SetOnMasterChangeFn(fn func(previous, current pb.ServerA mc.OnMasterChangeLock.Unlock() } +func (mc *MasterClient) SetOnVolumeLocationChangeFn(fn func(loc Location, addedVids, deletedVids []uint32)) { + mc.OnVolumeLocationChangeLock.Lock() + mc.OnVolumeLocationChange = fn + mc.OnVolumeLocationChangeLock.Unlock() +} + +func (mc *MasterClient) SetOnVidMapResetFn(fn func()) { + mc.OnVidMapResetLock.Lock() + mc.OnVidMapReset = fn + mc.OnVidMapResetLock.Unlock() +} + +// resetVidMap starts a new vid map generation and tells OnVidMapReset. +func (mc *MasterClient) resetVidMap() { + mc.vidMapClient.resetVidMap() + mc.OnVidMapResetLock.RLock() + fn := mc.OnVidMapReset + mc.OnVidMapResetLock.RUnlock() + if fn != nil { + fn() + } +} + func (mc *MasterClient) tryAllMasters(ctx context.Context) { var nextHintedLeader pb.ServerAddress failedMasters := make(map[pb.ServerAddress]struct{}) @@ -589,6 +620,7 @@ func (mc *MasterClient) updateVidMap(resp *master_pb.KeepConnectedResponse) { glog.V(2).Infof("%s.%s: %s masterClient removes ec volume %d", mc.FilerGroup, mc.clientType, loc.Url, deletedEcVid) mc.deleteEcLocation(deletedEcVid, loc) } + mc.notifyVolumeLocationChange(loc, resp.VolumeLocation) glog.V(1).Infof("updateVidMap(%s) %s.%s: %s volume add local: %d, remote: %d, del: %d, add ec: %d del ec: %d", resp.VolumeLocation.DataCenter, mc.FilerGroup, mc.clientType, loc.Url, len(resp.VolumeLocation.NewVids)-len(resp.VolumeLocation.RemoteVids), @@ -597,6 +629,32 @@ func (mc *MasterClient) updateVidMap(resp *master_pb.KeepConnectedResponse) { len(resp.VolumeLocation.DeletedEcVids)) } +// notifyVolumeLocationChange reports a removal only once loc serves the vid in +// no form, since ec.encode and ec.decode add the new form before removing the +// old one from the same server. +func (mc *MasterClient) notifyVolumeLocationChange(loc Location, vl *master_pb.VolumeLocation) { + mc.OnVolumeLocationChangeLock.RLock() + fn := mc.OnVolumeLocationChange + mc.OnVolumeLocationChangeLock.RUnlock() + if fn == nil { + return + } + var added, deleted []uint32 + for _, vids := range [][]uint32{vl.NewVids, vl.RemoteVids, vl.NewEcVids} { + added = append(added, vids...) + } + for _, vids := range [][]uint32{vl.DeletedVids, vl.DeletedEcVids} { + for _, vid := range vids { + if !mc.vidMap.hasLocation(vid, loc.Url) { + deleted = append(deleted, vid) + } + } + } + if len(added) > 0 || len(deleted) > 0 { + fn(loc, added, deleted) + } +} + func (mc *MasterClient) WithClient(ctx context.Context, streamingMode bool, fn func(client master_pb.SeaweedClient) error) error { getMasterF := func() pb.ServerAddress { return mc.GetMaster(ctx) diff --git a/weed/wdclient/masterclient_move_test.go b/weed/wdclient/masterclient_move_test.go index 9e77af4a13..c81e9ad411 100644 --- a/weed/wdclient/masterclient_move_test.go +++ b/weed/wdclient/masterclient_move_test.go @@ -91,3 +91,90 @@ func TestEcVolumeRemovedLosesItsLocation(t *testing.T) { t.Errorf("an ec volume that left the server kept its location: %v", locations) } } + +func listenForChanges(mc *MasterClient) (adds, dels *[]uint32) { + adds, dels = new([]uint32), new([]uint32) + mc.SetOnVolumeLocationChangeFn(func(_ Location, addedVids, deletedVids []uint32) { + *adds = append(*adds, addedVids...) + *dels = append(*dels, deletedVids...) + }) + return adds, dels +} + +func pushVolumeLocation(mc *MasterClient, url string, newVids, deletedVids, newEcVids, deletedEcVids []uint32) { + mc.updateVidMap(&master_pb.KeepConnectedResponse{VolumeLocation: &master_pb.VolumeLocation{ + Url: url, PublicUrl: url, NewVids: newVids, DeletedVids: deletedVids, NewEcVids: newEcVids, DeletedEcVids: deletedEcVids, + }}) +} + +// A volume moved between the server's disks arrives as an add, never a removal. +func TestChangeListenerHearsAppliedChanges(t *testing.T) { + mc := moveClient() + adds, dels := listenForChanges(mc) + + mc.updateVidMap(moveResponse([]uint32{1}, []uint32{1, 2})) + + if len(*adds) != 1 || (*adds)[0] != 1 { + t.Errorf("listener heard adds %v, want [1]", *adds) + } + if len(*dels) != 1 || (*dels)[0] != 2 { + t.Errorf("listener heard removals %v, want [2]", *dels) + } +} + +func TestChangeListenerSkipsRemovalWhileOtherFormServes(t *testing.T) { + mc := moveClient() + _, dels := listenForChanges(mc) + + pushVolumeLocation(mc, "server:8080", []uint32{1}, nil, []uint32{1}, nil) + pushVolumeLocation(mc, "server:8080", nil, nil, nil, []uint32{1}) + if len(*dels) != 0 { + t.Fatalf("shard removal reported while the regular copy stays: %v", *dels) + } + pushVolumeLocation(mc, "server:8080", nil, []uint32{1}, nil, nil) + if len(*dels) != 1 || (*dels)[0] != 1 { + t.Errorf("last form removed but listener heard %v, want [1]", *dels) + } +} + +// The shards and the regular copy's removal can share one message, where +// the removal is applied before the shards are added. +func TestChangeListenerSkipsRemovalReplacedInSameMessage(t *testing.T) { + mc := moveClient() + pushVolumeLocation(mc, "server:8080", []uint32{1}, nil, nil, nil) + _, dels := listenForChanges(mc) + + pushVolumeLocation(mc, "server:8080", nil, []uint32{1}, []uint32{1}, nil) + + if len(*dels) != 0 { + t.Errorf("regular copy replaced by shards in one message reported removed: %v", *dels) + } +} + +// Shards only an earlier master generation reported do not keep a server +// alive once the regular copy goes. +func TestChangeListenerIgnoresStaleGenerationWhenSuppressing(t *testing.T) { + mc := moveClient() + pushVolumeLocation(mc, "server:8080", []uint32{1}, nil, []uint32{1}, nil) + mc.resetVidMap() + pushVolumeLocation(mc, "server:8080", []uint32{1}, nil, nil, nil) + _, dels := listenForChanges(mc) + + pushVolumeLocation(mc, "server:8080", nil, []uint32{1}, nil, nil) + + if len(*dels) != 1 || (*dels)[0] != 1 { + t.Errorf("listener heard removals %v, want [1]", *dels) + } +} + +func TestResetVidMapNotifiesListener(t *testing.T) { + mc := moveClient() + resets := 0 + mc.SetOnVidMapResetFn(func() { resets++ }) + + mc.resetVidMap() + + if resets != 1 { + t.Errorf("reset listener heard %d resets, want 1", resets) + } +} diff --git a/weed/wdclient/vid_map.go b/weed/wdclient/vid_map.go index 944f683296..67fa4c83ae 100644 --- a/weed/wdclient/vid_map.go +++ b/weed/wdclient/vid_map.go @@ -206,6 +206,25 @@ func (vc *vidMap) GetLocations(vid uint32) (locations []Location, found bool) { return nil, false } +// hasLocation reports whether url serves vid, as a regular copy or ec shards, +// in an entry learned since the last reset. +func (vc *vidMap) hasLocation(vid uint32, url string) bool { + vc.RLock() + defer vc.RUnlock() + for _, vid2Locations := range []map[uint32]*locationsEntry{vc.vid2Locations, vc.ecVid2Locations} { + entry, found := vid2Locations[vid] + if !found || entry.generation != vc.generation { + continue + } + for _, loc := range entry.locations { + if loc.Url == url { + return true + } + } + } + return false +} + // lookupEntry returns vid's entry when it still holds locations. Callers must // hold the lock. func lookupEntry(vid2Locations map[uint32]*locationsEntry, vid uint32) (*locationsEntry, bool) { @@ -368,6 +387,16 @@ func (vc *vidMap) deleteVid(vid uint32) { } } +// clear drops every location, keeping no history to fall back to. +func (vc *vidMap) clear() { + vc.Lock() + defer vc.Unlock() + + vc.vid2Locations = make(map[uint32]*locationsEntry) + vc.ecVid2Locations = make(map[uint32]*locationsEntry) + vc.serverRefCount = make(map[string]int) +} + // reset starts a new generation, as when the master changes and everything it // told us has to be relearned. Entries stay readable while they are relearned // and are dropped once they fall out of the retained window. diff --git a/weed/wdclient/vidmap_client.go b/weed/wdclient/vidmap_client.go index 8ca713edc9..2cc71d949c 100644 --- a/weed/wdclient/vidmap_client.go +++ b/weed/wdclient/vidmap_client.go @@ -1,6 +1,7 @@ package wdclient import ( + "container/list" "context" "errors" "fmt" @@ -8,6 +9,8 @@ import ( "sort" "strconv" "strings" + "sync" + "time" "golang.org/x/sync/singleflight" @@ -30,8 +33,35 @@ type vidMapClient struct { vidMap *vidMap provider VolumeLocationProvider vidLookupGroup singleflight.Group + // recentDeletes fences (vid, url) pairs a pushed removal dropped, so a + // provider answer still naming them is neither cached nor served until the + // master reports the pair back or recentDeleteTTL passes. vidMapEpoch + // keeps answers fetched before a reset or clear out of the cache. + recentDeletesMu sync.Mutex + recentDeletes map[vidLocationKey]*list.Element + recentDeleteOrder *list.List // of recentDelete, oldest first + vidMapEpoch uint64 } +type recentDelete struct { + key vidLocationKey + at time.Time +} + +type vidLocationKey struct { + vid uint32 + url string +} + +const ( + // recentDeleteTTL outlasts a lookup in flight when the removal landed. + // Expiring only falls back to learning by a failed read, while a mark kept + // until the re-add would hide a live server whose re-add the master + // dropped on a full client channel. + recentDeleteTTL = time.Minute + recentDeleteCap = 1 << 16 +) + const ( // DefaultVidMapCacheSize is the default number of resets a volume location // survives without being relearned. This provides cache history when @@ -200,12 +230,12 @@ func (vc *vidMapClient) LookupVolumeIdsWithFallback(ctx context.Context, volumeI // Query provider with batched volume IDs glog.V(2).Infof("Looking up %d volumes from provider: %v", len(stillNeedLookup), stillNeedLookup) + fillEpoch := vc.mapEpoch() providerResults, err := vc.provider.LookupVolumeIds(ctx, stillNeedLookup) if err != nil { return batchResult, fmt.Errorf("provider lookup failed: %w", err) } - // Update cache with results for vidString, locations := range providerResults { vid, err := strconv.ParseUint(vidString, 10, 32) if err != nil { @@ -213,12 +243,15 @@ func (vc *vidMapClient) LookupVolumeIdsWithFallback(ctx context.Context, volumeI continue } + var live []Location for _, loc := range locations { - vc.addLocation(uint32(vid), loc) + if vc.fillLocation(uint32(vid), loc, fillEpoch) { + live = append(live, loc) + } } - if len(locations) > 0 { - batchResult[vidString] = locations + if len(live) > 0 { + batchResult[vidString] = live } } @@ -323,9 +356,89 @@ func (vc *vidMapClient) deleteEcLocation(vid uint32, location Location) { // resetVidMap starts a new generation, as when the master changes: what the // previous one told us stays readable until it is relearned or expires. func (vc *vidMapClient) resetVidMap() { + vc.recentDeletesMu.Lock() + vc.vidMapEpoch++ + vc.recentDeletesMu.Unlock() vc.vidMap.reset() } +// clearVidMap drops every location, for a cache whose feed restarted without +// replaying the removals it missed. +func (vc *vidMapClient) clearVidMap() { + vc.recentDeletesMu.Lock() + vc.vidMapEpoch++ + vc.recentDeletesMu.Unlock() + vc.vidMap.clear() +} + +func (vc *vidMapClient) mapEpoch() uint64 { + vc.recentDeletesMu.Lock() + defer vc.recentDeletesMu.Unlock() + return vc.vidMapEpoch +} + +func (vc *vidMapClient) tombstone(vid uint32, loc Location) { + key := vidLocationKey{vid: vid, url: loc.Url} + now := time.Now() + vc.recentDeletesMu.Lock() + defer vc.recentDeletesMu.Unlock() + if vc.recentDeletes == nil { + vc.recentDeletes = make(map[vidLocationKey]*list.Element) + vc.recentDeleteOrder = list.New() + } + vc.expireRecentDeletes(now) + if e, found := vc.recentDeletes[key]; found { + e.Value = recentDelete{key: key, at: now} + vc.recentDeleteOrder.MoveToBack(e) + return + } + vc.recentDeletes[key] = vc.recentDeleteOrder.PushBack(recentDelete{key: key, at: now}) + if len(vc.recentDeletes) > recentDeleteCap { + vc.removeRecentDelete(vc.recentDeleteOrder.Front()) + } +} + +// expireRecentDeletes drops marks older than recentDeleteTTL. Callers must +// hold recentDeletesMu. +func (vc *vidMapClient) expireRecentDeletes(now time.Time) { + if vc.recentDeleteOrder == nil { + return + } + for e := vc.recentDeleteOrder.Front(); e != nil && now.Sub(e.Value.(recentDelete).at) >= recentDeleteTTL; e = vc.recentDeleteOrder.Front() { + vc.removeRecentDelete(e) + } +} + +func (vc *vidMapClient) removeRecentDelete(e *list.Element) { + delete(vc.recentDeletes, e.Value.(recentDelete).key) + vc.recentDeleteOrder.Remove(e) +} + +func (vc *vidMapClient) untombstone(vid uint32, loc Location) { + key := vidLocationKey{vid: vid, url: loc.Url} + vc.recentDeletesMu.Lock() + defer vc.recentDeletesMu.Unlock() + if e, found := vc.recentDeletes[key]; found { + vc.removeRecentDelete(e) + } +} + +// fillLocation reports whether a provider answer may be served, and caches it +// unless a reset or clear happened since epoch. Holding recentDeletesMu across +// the check and the add keeps a concurrent removal from being undone. +func (vc *vidMapClient) fillLocation(vid uint32, loc Location, epoch uint64) bool { + vc.recentDeletesMu.Lock() + defer vc.recentDeletesMu.Unlock() + vc.expireRecentDeletes(time.Now()) + if _, found := vc.recentDeletes[vidLocationKey{vid: vid, url: loc.Url}]; found { + return false + } + if vc.vidMapEpoch == epoch { + vc.vidMap.addLocation(vid, loc) + } + return true +} + // InvalidateCache removes all cached locations for a volume ID func (vc *vidMapClient) InvalidateCache(fileId string) { parts := strings.Split(fileId, ",") diff --git a/weed/wdclient/vidmap_client_localfirst_test.go b/weed/wdclient/vidmap_client_localfirst_test.go index 931fe5e4b2..23a536fbfe 100644 --- a/weed/wdclient/vidmap_client_localfirst_test.go +++ b/weed/wdclient/vidmap_client_localfirst_test.go @@ -4,6 +4,7 @@ import ( "context" "strings" "testing" + "time" ) type testLocationProvider struct { @@ -111,3 +112,85 @@ func TestLookupFileIdWithFallbackKeepsDataCenterFirst(t *testing.T) { } } } + +// resetOnLookupProvider answers after clearing the cache, as a provider +// response landing on the far side of a master reconnect would. +type resetOnLookupProvider struct { + vc *vidMapClient +} + +func (p *resetOnLookupProvider) LookupVolumeIds(ctx context.Context, volumeIds []string) (map[string][]Location, error) { + p.vc.clearVidMap() + return map[string][]Location{"5": {{Url: "stale:8080"}}}, nil +} + +func TestProviderFillServesWithoutCachingAcrossReset(t *testing.T) { + provider := &resetOnLookupProvider{} + vc := newVidMapClient(provider, "", 5) + provider.vc = vc + + locations, err := vc.LookupVolumeIdsWithFallback(context.Background(), []string{"5"}) + if err != nil || len(locations["5"]) != 1 { + t.Fatalf("lookup crossing a cache reset returned %v, err %v", locations["5"], err) + } + if cached, found := vc.GetLocations(5); found && len(cached) > 0 { + t.Fatalf("lookup crossed a cache reset but cached %v", cached) + } +} + +func TestProviderFillSkipsTombstonedLocation(t *testing.T) { + retired := Location{Url: "retired:8080"} + live := Location{Url: "live:8080"} + vc := newVidMapClient(&testLocationProvider{ + locations: map[string][]Location{"5": {retired, live}}, + }, "", 5) + + vc.tombstone(5, retired) + + locations, err := vc.LookupVolumeIdsWithFallback(context.Background(), []string{"5"}) + if err != nil { + t.Fatalf("lookup failed: %v", err) + } + if got := locations["5"]; len(got) != 1 || got[0].Url != live.Url { + t.Fatalf("lookup returned %v, want only the surviving location", got) + } + if cached, found := vc.GetLocations(5); !found || len(cached) != 1 || cached[0].Url != live.Url { + t.Errorf("cache holds %v, want only the surviving location", cached) + } +} + +func TestTombstoneExpires(t *testing.T) { + retired := Location{Url: "retired:8080"} + vc := newVidMapClient(&testLocationProvider{ + locations: map[string][]Location{"5": {retired}}, + }, "", 5) + vc.tombstone(5, retired) + e := vc.recentDeletes[vidLocationKey{vid: 5, url: retired.Url}] + e.Value = recentDelete{key: vidLocationKey{vid: 5, url: retired.Url}, at: time.Now().Add(-recentDeleteTTL)} + + locations, err := vc.LookupVolumeIdsWithFallback(context.Background(), []string{"5"}) + if err != nil || len(locations["5"]) != 1 { + t.Fatalf("expired mark still fenced the location: %v, err %v", locations["5"], err) + } + if len(vc.recentDeletes) != 0 || vc.recentDeleteOrder.Len() != 0 { + t.Errorf("expired mark kept: %d marks, %d ordered", len(vc.recentDeletes), vc.recentDeleteOrder.Len()) + } +} + +func TestTombstoneEvictsOldestAtCap(t *testing.T) { + vc := newVidMapClient(nil, "", 5) + for vid := uint32(0); vid <= recentDeleteCap; vid++ { + vc.tombstone(vid, Location{Url: "a:8080"}) + } + vc.tombstone(recentDeleteCap, Location{Url: "a:8080"}) + + if len(vc.recentDeletes) != recentDeleteCap || vc.recentDeleteOrder.Len() != recentDeleteCap { + t.Fatalf("fence holds %d marks (%d ordered), want %d", len(vc.recentDeletes), vc.recentDeleteOrder.Len(), recentDeleteCap) + } + if _, found := vc.recentDeletes[vidLocationKey{vid: 0, url: "a:8080"}]; found { + t.Errorf("oldest mark survived eviction") + } + if _, found := vc.recentDeletes[vidLocationKey{vid: 1, url: "a:8080"}]; !found { + t.Errorf("a repeated mark evicted another one") + } +}