From b09bd2abcbb0a04ffae11fc017dbe7f23a5eb058 Mon Sep 17 00:00:00 2001 From: xiejian Date: Thu, 23 Apr 2026 15:27:10 +0800 Subject: [PATCH] fix(proxy): handle current/backup swap during retain with: #1000910244 Signed-off-by: xiejian --- .../proxy/allocator/retain_volume_task.go | 22 +++++----- blobstore/proxy/allocator/volume_test.go | 43 +++++++++++++++++++ blobstore/proxy/allocator/volumemgr.go | 11 +++++ 3 files changed, 65 insertions(+), 11 deletions(-) diff --git a/blobstore/proxy/allocator/retain_volume_task.go b/blobstore/proxy/allocator/retain_volume_task.go index d3e3dffcd..ad79dfd9b 100644 --- a/blobstore/proxy/allocator/retain_volume_task.go +++ b/blobstore/proxy/allocator/retain_volume_task.go @@ -81,7 +81,7 @@ func (v *volumeMgr) retain(ctx context.Context, isBackup bool) { if end > numTokens { end = numTokens } - v.retainVolumeFromCm(ctx, retainTokenArgs[start:end], isBackup) + v.retainVolumeFromCm(ctx, retainTokenArgs[start:end]) start = end if start < numTokens { time.Sleep(time.Duration(v.RetainBatchIntervalS) * time.Second) @@ -117,12 +117,12 @@ func (v *volumeMgr) handleFullVols(ctx context.Context, isBackup bool) { // remove retain failed volume and update success expire time func (v *volumeMgr) handleRetainResult(ctx context.Context, retainTokenArgs []string, - retainRet []clustermgr.RetainVolume, isBackup bool, + retainRet []clustermgr.RetainVolume, ) { span := trace.SpanFromContextSafe(ctx) if len(retainRet) == 0 { for _, token := range retainTokenArgs { - err := v.discardVolume(ctx, token, isBackup) + err := v.discardVolume(ctx, token) if err != nil { span.Error(err) } @@ -136,7 +136,7 @@ func (v *volumeMgr) handleRetainResult(ctx context.Context, retainTokenArgs []st span.Errorf("decodeToken %v error", infos.Token) continue } - err = v.updateExpireTime(vid, infos.ExpireTime, isBackup) + err = v.updateExpireTime(vid, infos.ExpireTime) if err != nil { span.Error(err, vid) } @@ -146,14 +146,14 @@ func (v *volumeMgr) handleRetainResult(ctx context.Context, retainTokenArgs []st if _, ok := tokenMap[token]; ok { continue } - err := v.discardVolume(ctx, token, isBackup) + err := v.discardVolume(ctx, token) if err != nil { span.Error(err, token) } } } -func (v *volumeMgr) discardVolume(ctx context.Context, token string, isBackup bool) (err error) { +func (v *volumeMgr) discardVolume(ctx context.Context, token string) (err error) { span := trace.SpanFromContextSafe(ctx) _, vid, err := proto.DecodeToken(token) if err != nil { @@ -161,7 +161,7 @@ func (v *volumeMgr) discardVolume(ctx context.Context, token string, isBackup bo } span.Infof("retain failed vid: %v", vid) for _, info := range v.modeInfos { - if vol, ok := info.Get(vid, isBackup); ok { + if vol, ok := info.GetAny(vid); ok { vol.mu.Lock() vol.deleted = true vol.mu.Unlock() @@ -190,9 +190,9 @@ func (v *volumeMgr) genRetainVolume(ctx context.Context, isBackup bool) (tokens return } -func (v *volumeMgr) updateExpireTime(vid proto.Vid, expireTime int64, isBackup bool) (err error) { +func (v *volumeMgr) updateExpireTime(vid proto.Vid, expireTime int64) (err error) { for _, modeInfo := range v.modeInfos { - if vol, ok := modeInfo.Get(vid, isBackup); ok { + if vol, ok := modeInfo.GetAny(vid); ok { vol.ExpireTime = expireTime return nil } @@ -200,7 +200,7 @@ func (v *volumeMgr) updateExpireTime(vid proto.Vid, expireTime int64, isBackup b return errors.New("vid does not exist ") } -func (v *volumeMgr) retainVolumeFromCm(ctx context.Context, tokens []string, isBackup bool) { +func (v *volumeMgr) retainVolumeFromCm(ctx context.Context, tokens []string) { span := trace.SpanFromContextSafe(ctx) args := &clustermgr.RetainVolumeArgs{ Tokens: tokens, @@ -211,7 +211,7 @@ func (v *volumeMgr) retainVolumeFromCm(ctx context.Context, tokens []string, isB return } span.Debugf("retain result: %#v, lens: %v\n", retainVolume, len(retainVolume.RetainVolTokens)) - v.handleRetainResult(ctx, tokens, retainVolume.RetainVolTokens, isBackup) + v.handleRetainResult(ctx, tokens, retainVolume.RetainVolTokens) } // checkAndReplenish inspects the volume cache for every code mode after a retain cycle. diff --git a/blobstore/proxy/allocator/volume_test.go b/blobstore/proxy/allocator/volume_test.go index 3058251bd..41ee4c996 100644 --- a/blobstore/proxy/allocator/volume_test.go +++ b/blobstore/proxy/allocator/volume_test.go @@ -89,6 +89,49 @@ func TestVolumes_Put(t *testing.T) { require.False(t, ok) } +func TestModeInfo_GetAny(t *testing.T) { + newVol := func(vid proto.Vid) *volume { + return &volume{AllocVolumeInfo: cm.AllocVolumeInfo{ + VolumeInfo: cm.VolumeInfo{ + VolumeInfoBase: cm.VolumeInfoBase{ + Free: 10 * 16 * 1024 * 1024 * 1024, + Vid: vid, + }, + }, + ExpireTime: 100, + }} + } + + m := &modeInfo{current: &volumes{}, backup: &volumes{}} + m.Put(newVol(1), false) + m.Put(newVol(2), true) + + vol, ok := m.GetAny(proto.Vid(1)) + require.True(t, ok) + require.Equal(t, proto.Vid(1), vol.Vid) + + vol, ok = m.GetAny(proto.Vid(2)) + require.True(t, ok) + require.Equal(t, proto.Vid(2), vol.Vid) + + _, ok = m.GetAny(proto.Vid(99)) + require.False(t, ok) + + // simulate current/backup swap during a retain RPC: the vol that used to + // be in current is now in backup; GetAny should still find it. + m.lock.Lock() + m.current, m.backup = m.backup, m.current + m.lock.Unlock() + + vol, ok = m.GetAny(proto.Vid(1)) + require.True(t, ok) + require.Equal(t, proto.Vid(1), vol.Vid) + + vol, ok = m.GetAny(proto.Vid(2)) + require.True(t, ok) + require.Equal(t, proto.Vid(2), vol.Vid) +} + func TestVolumes_Delete(t *testing.T) { v := volumes{} for _, i := range []int{1, 3, 2, 4} { diff --git a/blobstore/proxy/allocator/volumemgr.go b/blobstore/proxy/allocator/volumemgr.go index 31d8104a5..8a4a54ac2 100644 --- a/blobstore/proxy/allocator/volumemgr.go +++ b/blobstore/proxy/allocator/volumemgr.go @@ -141,6 +141,17 @@ func (m *modeInfo) Get(vid proto.Vid, isBackup bool) (res *volume, ok bool) { return } +func (m *modeInfo) GetAny(vid proto.Vid) (res *volume, ok bool) { + m.lock.RLock() + if res, ok = m.current.Get(vid); ok { + m.lock.RUnlock() + return + } + res, ok = m.backup.Get(vid) + m.lock.RUnlock() + return +} + func (m *modeInfo) TotalFree() int64 { m.lock.RLock() totalFree := m.backup.TotalFree() + m.current.TotalFree()