[#xx] engine: Add tests for object deletion on read-only shard
Some checks failed
DCO action / DCO (pull_request) Failing after 1m9s
Tests and linters / Run gofumpt (pull_request) Successful in 2m26s
Vulncheck / Vulncheck (pull_request) Successful in 2m35s
Pre-commit hooks / Pre-commit (pull_request) Successful in 2m41s
Build / Build Components (pull_request) Successful in 3m6s
Tests and linters / Lint (pull_request) Successful in 3m47s
Tests and linters / Staticcheck (pull_request) Successful in 3m46s
Tests and linters / gopls check (pull_request) Successful in 3m48s
Tests and linters / Tests (pull_request) Failing after 6m8s
Tests and linters / Tests with -race (pull_request) Failing after 6m11s

Currently, when an object on some shard is inhumed, the engine places a
tombstone on the same shard. If the target shard is read-only, the engine fails.

In that case, we want the engine to correctly place a tombstone on a different
shard, ensuring that:
- An object becomes unavailable if a tombstone was placed on a different shard.
  See `TestObjectUnavailableIfTombstoneOnDifferentShard` test.
- GC deletes an object if a tombstone was placed on a different shard.
  See `TestObjectDeletedIfTombstoneOnDifferentShard` test.

Signed-off-by: Aleksey Savchuk <a.savchuk@yadro.com>
This commit is contained in:
Aleksey Savchuk 2024-10-01 12:33:27 +03:00
parent 61d5e140e0
commit 0c15185883
No known key found for this signature in database

View file

@ -3,12 +3,17 @@ package engine
import (
"context"
"testing"
"time"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/testutil"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
cidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id/test"
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
oidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id/test"
objecttest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/test"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@ -83,3 +88,116 @@ func TestStorageEngine_Inhume(t *testing.T) {
require.Empty(t, addrs)
})
}
func TestObjectUnavailableIfTombstoneOnDifferentShard(t *testing.T) {
t.Run("look first at the shard with the object", func(t *testing.T) {
obj := objecttest.Object().Parent()
shard1 := testNewShard(t, shard.WithDisabledGC())
shard2 := testNewShard(t, shard.WithDisabledGC())
engine := testNewEngine(t).setInitializedShards(t, shard1, shard2).engine
shards := engine.sortShards(object.AddressOf(obj))
shard1 = shards[0].Shard
shard2 = shards[1].Shard
testObjectUnavailableIfTombstoneOnDifferentShard(t, obj, shard1, shard2, engine)
})
t.Run("look first at the shard with the tombstone", func(t *testing.T) {
obj := objecttest.Object().Parent()
shard1 := testNewShard(t, shard.WithDisabledGC())
shard2 := testNewShard(t, shard.WithDisabledGC())
engine := testNewEngine(t).setInitializedShards(t, shard1, shard2).engine
shards := engine.sortShards(object.AddressOf(obj))
shard1 = shards[1].Shard
shard2 = shards[0].Shard
testObjectUnavailableIfTombstoneOnDifferentShard(t, obj, shard1, shard2, engine)
})
}
func testObjectUnavailableIfTombstoneOnDifferentShard(
t *testing.T,
obj *objectSDK.Object,
shard1, shard2 *shard.Shard,
engine *StorageEngine,
) {
ctx := context.Background()
ts := oidtest.Address()
cnr, set := obj.ContainerID()
require.True(t, set)
ts.SetContainer(cnr)
{
prm := shard.PutPrm{}
prm.SetObject(obj)
_, err := shard1.Put(ctx, prm)
require.NoError(t, err)
}
{
prm := shard.InhumePrm{}
prm.SetTarget(ts, object.AddressOf(obj))
_, err := shard2.Inhume(ctx, prm)
require.NoError(t, err)
}
{
prm := GetPrm{}
prm.WithAddress(object.AddressOf(obj))
_, err := engine.Get(ctx, prm)
assert.Error(t, err)
assert.True(t, client.IsErrObjectAlreadyRemoved(err))
}
}
func TestObjectDeletedIfTombstoneOnDifferentShard(t *testing.T) {
ctx := context.Background()
shard1 := testNewShard(t, shard.WithGCRemoverSleepInterval(200*time.Millisecond))
shard2 := testNewShard(t, shard.WithGCRemoverSleepInterval(200*time.Millisecond))
obj := objecttest.Object().Parent()
ts := oidtest.Address()
cnr, set := obj.ContainerID()
require.True(t, set)
ts.SetContainer(cnr)
{
prm := shard.PutPrm{}
prm.SetObject(obj)
_, err := shard1.Put(ctx, prm)
require.NoError(t, err)
}
{
prm := shard.InhumePrm{}
prm.SetTarget(ts, object.AddressOf(obj))
_, err := shard2.Inhume(ctx, prm)
require.NoError(t, err)
}
<-time.After(1 * time.Second)
{
prm := shard.ExistsPrm{
Address: object.AddressOf(obj),
}
res, err := shard1.Exists(ctx, prm)
assert.NoError(t, err)
assert.False(t, res.Exists())
_, err = shard2.Exists(ctx, prm)
require.True(t, client.IsErrObjectAlreadyRemoved(err))
}
}