diff --git a/pkg/local_object_storage/blobstor/blobovniczatree/rebuild.go b/pkg/local_object_storage/blobstor/blobovniczatree/rebuild.go index b7f20822e..b7b1dfd4b 100644 --- a/pkg/local_object_storage/blobstor/blobovniczatree/rebuild.go +++ b/pkg/local_object_storage/blobstor/blobovniczatree/rebuild.go @@ -172,7 +172,7 @@ func (b *Blobovniczas) selectDBsDoNotMatchFillPercent(ctx context.Context, targe continue } path := filepath.Join(lvlPath, e.Name()) - resettlementRequired, err := b.fillPercentIsLow(path, target) + resettlementRequired, err := b.rebuildBySize(path, target) if err != nil { return false, err } @@ -187,14 +187,19 @@ func (b *Blobovniczas) selectDBsDoNotMatchFillPercent(ctx context.Context, targe return result, nil } -func (b *Blobovniczas) fillPercentIsLow(path string, target int) (bool, error) { +func (b *Blobovniczas) rebuildBySize(path string, targetFillPercent int) (bool, error) { shDB := b.getBlobovnicza(path) blz, err := shDB.Open() if err != nil { return false, err } defer shDB.Close() - return blz.FillPercent() < target, nil + fp := blz.FillPercent() + // accepted fill percent defines as + // |----|+++++++++++++++++|+++++++++++++++++|--------------- + // 0% target 100% 100+(100 - target) + // where `+` - accepted fill percent, `-` - not accepted fill percent + return fp < targetFillPercent || fp > 100+(100-targetFillPercent), nil } func (b *Blobovniczas) rebuildDB(ctx context.Context, path string, meta common.MetaStorage, limiter common.ConcurrentWorkersLimiter) (uint64, error) { diff --git a/pkg/local_object_storage/blobstor/blobovniczatree/rebuild_test.go b/pkg/local_object_storage/blobstor/blobovniczatree/rebuild_test.go index 62ae9ea90..e6da1c553 100644 --- a/pkg/local_object_storage/blobstor/blobovniczatree/rebuild_test.go +++ b/pkg/local_object_storage/blobstor/blobovniczatree/rebuild_test.go @@ -228,6 +228,80 @@ func TestBlobovniczaTreeFillPercentRebuild(t *testing.T) { require.NoError(t, b.Close()) }) + + t.Run("rebuild by overflow", func(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + b := NewBlobovniczaTree( + context.Background(), + WithLogger(test.NewLogger(t)), + WithObjectSizeLimit(64*1024), + WithBlobovniczaShallowWidth(1), // single directory + WithBlobovniczaShallowDepth(1), + WithRootPath(dir), + WithBlobovniczaSize(100*1024), // 100 KB limit for each blobovnicza + WithWaitBeforeDropDB(0), + WithOpenedCacheSize(1000), + WithMoveBatchSize(3)) + require.NoError(t, b.Open(mode.ComponentReadWrite)) + require.NoError(t, b.Init()) + + storageIDs := make(map[oid.Address][]byte) + for i := 0; i < 100; i++ { // 2 objects for one blobovnicza, so 50 DBs total will be created + obj := blobstortest.NewObject(64 * 1024) + data, err := obj.Marshal() + require.NoError(t, err) + var prm common.PutPrm + prm.Address = object.AddressOf(obj) + prm.RawData = data + res, err := b.Put(context.Background(), prm) + require.NoError(t, err) + storageIDs[prm.Address] = res.StorageID + } + metaStub := &storageIDUpdateStub{ + storageIDs: storageIDs, + guard: &sync.Mutex{}, + } + require.NoError(t, b.Close()) + b = NewBlobovniczaTree( + context.Background(), + WithLogger(test.NewLogger(t)), + WithObjectSizeLimit(64*1024), + WithBlobovniczaShallowWidth(1), + WithBlobovniczaShallowDepth(1), + WithRootPath(dir), + WithBlobovniczaSize(50*1024), // 50 KB limit for each blobovnicza + WithWaitBeforeDropDB(0), + WithOpenedCacheSize(1000), + WithMoveBatchSize(3)) + require.NoError(t, b.Open(mode.ComponentReadWrite)) + require.NoError(t, b.Init()) + + rRes, err := b.Rebuild(context.Background(), common.RebuildPrm{ + MetaStorage: metaStub, + WorkerLimiter: &rebuildLimiterStub{}, + Action: common.RebuildAction{ + SchemaChange: false, + FillPercent: true, + FillPercentValue: 80, + }, + }) + require.NoError(t, err) + require.Equal(t, uint64(49), rRes.FilesRemoved) + require.Equal(t, uint64(98), rRes.ObjectsMoved) // 49 DBs with 2 objects + require.Equal(t, uint64(98), metaStub.updatedCount) + + for addr, storageID := range storageIDs { + var gPrm common.GetPrm + gPrm.Address = addr + gPrm.StorageID = storageID + _, err := b.Get(context.Background(), gPrm) + require.NoError(t, err) + } + + require.NoError(t, b.Close()) + }) } func TestBlobovniczaTreeRebuildLargeObject(t *testing.T) {