[#1403] engine: Add tests for object deletion on read-only shard

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>

pkg/local_object_storage/engine/inhume_test.go

@@ -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))
+	}
+}