[#1563] tree: Wrap only ChainRouterError erros with ObjectAccessDenied

* Such wrapping helps to differentiate logical check errors and server internal
  errors.

Signed-off-by: Airat Arifullin <a.arifullin@yadro.com>

cmd/frostfs-cli/modules/object/nodes.go

@@ -1,15 +1,12 @@
 package object
 
 import (
-	"bytes"
-	"cmp"
 	"context"
 	"crypto/ecdsa"
 	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
-	"slices"
 	"sync"
 
 	internalclient "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-cli/internal/client"
@@ -507,7 +504,6 @@ func isObjectStoredOnNode(ctx context.Context, cmd *cobra.Command, cnrID cid.ID,
 }
 
 func printPlacement(cmd *cobra.Command, objID oid.ID, objects []phyObject, result *objectNodesResult) {
-	normilizeObjectNodesResult(objects, result)
 	if json, _ := cmd.Flags().GetBool(commonflags.JSON); json {
 		printObjectNodesAsJSON(cmd, objID, objects, result)
 	} else {
@@ -515,34 +511,6 @@ func printPlacement(cmd *cobra.Command, objID oid.ID, objects []phyObject, resul
 	}
 }
 
-func normilizeObjectNodesResult(objects []phyObject, result *objectNodesResult) {
-	slices.SortFunc(objects, func(lhs, rhs phyObject) int {
-		if lhs.ecHeader == nil && rhs.ecHeader == nil {
-			return bytes.Compare(lhs.objectID[:], rhs.objectID[:])
-		}
-		if lhs.ecHeader == nil {
-			return -1
-		}
-		if rhs.ecHeader == nil {
-			return 1
-		}
-		if lhs.ecHeader.parent == rhs.ecHeader.parent {
-			return cmp.Compare(lhs.ecHeader.index, rhs.ecHeader.index)
-		}
-		return bytes.Compare(lhs.ecHeader.parent[:], rhs.ecHeader.parent[:])
-	})
-	for _, obj := range objects {
-		op := result.placements[obj.objectID]
-		slices.SortFunc(op.confirmedNodes, func(lhs, rhs netmapSDK.NodeInfo) int {
-			return bytes.Compare(lhs.PublicKey(), rhs.PublicKey())
-		})
-		slices.SortFunc(op.requiredNodes, func(lhs, rhs netmapSDK.NodeInfo) int {
-			return bytes.Compare(lhs.PublicKey(), rhs.PublicKey())
-		})
-		result.placements[obj.objectID] = op
-	}
-}
-
 func printObjectNodesAsText(cmd *cobra.Command, objID oid.ID, objects []phyObject, result *objectNodesResult) {
 	fmt.Fprintf(cmd.OutOrStdout(), "Object %s stores payload in %d data objects:\n", objID.EncodeToString(), len(objects))
 

cmd/frostfs-node/config/engine/config.go

@@ -41,6 +41,10 @@ func IterateShards(c *config.Config, required bool, f func(*shardconfig.Config)
 			c.Sub(si),
 		)
 
+		if sc.Mode() == mode.Disabled {
+			continue
+		}
+
 		// Path for the blobstor can't be present in the default section, because different shards
 		// must have different paths, so if it is missing, the shard is not here.
 		// At the same time checking for "blobstor" section doesn't work proper
@@ -50,10 +54,6 @@ func IterateShards(c *config.Config, required bool, f func(*shardconfig.Config)
 		}
 		(*config.Config)(sc).SetDefault(def)
 
-		if sc.Mode() == mode.Disabled {
-			continue
-		}
-
 		if err := f(sc); err != nil {
 			return err
 		}

cmd/frostfs-node/config/engine/config_test.go

@@ -18,6 +18,22 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestIterateShards(t *testing.T) {
+	fileConfigTest := func(c *config.Config) {
+		var res []string
+		require.NoError(t,
+			engineconfig.IterateShards(c, false, func(sc *shardconfig.Config) error {
+				res = append(res, sc.Metabase().Path())
+				return nil
+			}))
+		require.Equal(t, []string{"abc", "xyz"}, res)
+	}
+
+	const cfgDir = "./testdata/shards"
+	configtest.ForEachFileType(cfgDir, fileConfigTest)
+	configtest.ForEnvFileType(t, cfgDir, fileConfigTest)
+}
+
 func TestEngineSection(t *testing.T) {
 	t.Run("defaults", func(t *testing.T) {
 		empty := configtest.EmptyConfig()

cmd/frostfs-node/config/engine/testdata/shards.env

@@ -0,0 +1,3 @@
+FROSTFS_STORAGE_SHARD_0_METABASE_PATH=abc
+FROSTFS_STORAGE_SHARD_1_MODE=disabled
+FROSTFS_STORAGE_SHARD_2_METABASE_PATH=xyz

cmd/frostfs-node/config/engine/testdata/shards.json

@@ -0,0 +1,13 @@
+{
+  "storage.shard": {
+    "0": {
+      "metabase.path": "abc"
+    },
+    "1": {
+      "mode": "disabled"
+    },
+    "2": {
+      "metabase.path": "xyz"
+    }
+  }
+}

cmd/frostfs-node/config/engine/testdata/shards.yaml

@@ -0,0 +1,7 @@
+storage.shard:
+  0:
+    metabase.path: abc
+  1:
+    mode: disabled
+  2:
+    metabase.path: xyz

go.mod

@@ -8,7 +8,7 @@ require (
 	git.frostfs.info/TrueCloudLab/frostfs-crypto v0.6.0
 	git.frostfs.info/TrueCloudLab/frostfs-locode-db v0.4.1-0.20240710074952-65761deb5c0d
 	git.frostfs.info/TrueCloudLab/frostfs-observability v0.0.0-20241112082307-f17779933e88
-	git.frostfs.info/TrueCloudLab/frostfs-sdk-go v0.0.0-20241107121119-cb813e27a823
+	git.frostfs.info/TrueCloudLab/frostfs-sdk-go v0.0.0-20241206094944-81c423e7094d
 	git.frostfs.info/TrueCloudLab/hrw v1.2.1
 	git.frostfs.info/TrueCloudLab/multinet v0.0.0-20241015075604-6cb0d80e0972
 	git.frostfs.info/TrueCloudLab/policy-engine v0.0.0-20240814080254-96225afacb88

go.sum

@@ -8,8 +8,8 @@ git.frostfs.info/TrueCloudLab/frostfs-locode-db v0.4.1-0.20240710074952-65761deb
 git.frostfs.info/TrueCloudLab/frostfs-locode-db v0.4.1-0.20240710074952-65761deb5c0d/go.mod h1:7ZZq8iguY7qFsXajdHGmZd2AW4QbucyrJwhbsRfOfek=
 git.frostfs.info/TrueCloudLab/frostfs-observability v0.0.0-20241112082307-f17779933e88 h1:9bvBDLApbbO5sXBKdODpE9tzy3HV99nXxkDWNn22rdI=
 git.frostfs.info/TrueCloudLab/frostfs-observability v0.0.0-20241112082307-f17779933e88/go.mod h1:kbwB4v2o6RyOfCo9kEFeUDZIX3LKhmS0yXPrtvzkQ1g=
-git.frostfs.info/TrueCloudLab/frostfs-sdk-go v0.0.0-20241107121119-cb813e27a823 h1:sepm9FeuoInmygH1K/+3L+Yp5bJhGiVi/oGCH6Emp2c=
-git.frostfs.info/TrueCloudLab/frostfs-sdk-go v0.0.0-20241107121119-cb813e27a823/go.mod h1:eoK7+KZQ9GJxbzIs6vTnoUJqFDppavInLRHaN4MYgZg=
+git.frostfs.info/TrueCloudLab/frostfs-sdk-go v0.0.0-20241206094944-81c423e7094d h1:FpXI+mOrmJk3t2MKQFZuhLjCHDyDeo5rtP1WXl7gUWc=
+git.frostfs.info/TrueCloudLab/frostfs-sdk-go v0.0.0-20241206094944-81c423e7094d/go.mod h1:eoK7+KZQ9GJxbzIs6vTnoUJqFDppavInLRHaN4MYgZg=
 git.frostfs.info/TrueCloudLab/hrw v1.2.1 h1:ccBRK21rFvY5R1WotI6LNoPlizk7qSvdfD8lNIRudVc=
 git.frostfs.info/TrueCloudLab/hrw v1.2.1/go.mod h1:C1Ygde2n843yTZEQ0FP69jYiuaYV0kriLvP4zm8JuvM=
 git.frostfs.info/TrueCloudLab/multinet v0.0.0-20241015075604-6cb0d80e0972 h1:/960fWeyn2AFHwQUwDsWB3sbP6lTEnFnMzLMM6tx6N8=

pkg/local_object_storage/engine/exists_test.go

@@ -42,7 +42,7 @@ func benchmarkExists(b *testing.B, shardNum int) {
 	for range b.N {
 		var shPrm shard.ExistsPrm
 		shPrm.Address = addr
-		shPrm.ParentAddress = oid.Address{}
+		shPrm.ECParentAddress = oid.Address{}
 		ok, _, err := e.exists(context.Background(), shPrm)
 		if err != nil || ok {
 			b.Fatalf("%t %v", ok, err)

pkg/local_object_storage/engine/inhume_test.go

@@ -7,8 +7,11 @@ import (
 	"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"
+	apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
 	cidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id/test"
 	objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
+	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
+	oidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id/test"
 	"github.com/stretchr/testify/require"
 )
 
@@ -84,3 +87,47 @@ func TestStorageEngine_Inhume(t *testing.T) {
 		require.Empty(t, addrs)
 	})
 }
+
+func TestStorageEngine_ECInhume(t *testing.T) {
+	parentObjectAddress := oidtest.Address()
+	containerID := parentObjectAddress.Container()
+
+	chunkObject0 := testutil.GenerateObjectWithCID(containerID)
+	chunkObject0.SetECHeader(objectSDK.NewECHeader(
+		objectSDK.ECParentInfo{
+			ID: parentObjectAddress.Object(),
+		}, 0, 4, []byte{}, 0))
+
+	chunkObject1 := testutil.GenerateObjectWithCID(containerID)
+	chunkObject1.SetECHeader(objectSDK.NewECHeader(
+		objectSDK.ECParentInfo{
+			ID: parentObjectAddress.Object(),
+		}, 1, 4, []byte{}, 0))
+
+	tombstone := objectSDK.NewTombstone()
+	tombstone.SetMembers([]oid.ID{parentObjectAddress.Object()})
+	payload, err := tombstone.Marshal()
+	require.NoError(t, err)
+	tombstoneObject := testutil.GenerateObjectWithCID(containerID)
+	tombstoneObject.SetType(objectSDK.TypeTombstone)
+	tombstoneObject.SetPayload(payload)
+	tombstoneObjectAddress := object.AddressOf(tombstoneObject)
+
+	e := testNewEngine(t).setShardsNum(t, 5).prepare(t).engine
+	defer func() { require.NoError(t, e.Close(context.Background())) }()
+
+	require.NoError(t, Put(context.Background(), e, chunkObject0, false))
+
+	require.NoError(t, Put(context.Background(), e, tombstoneObject, false))
+
+	var inhumePrm InhumePrm
+	inhumePrm.WithTarget(tombstoneObjectAddress, parentObjectAddress)
+	_, err = e.Inhume(context.Background(), inhumePrm)
+	require.NoError(t, err)
+
+	var alreadyRemoved *apistatus.ObjectAlreadyRemoved
+
+	require.ErrorAs(t, Put(context.Background(), e, chunkObject0, false), &alreadyRemoved)
+
+	require.ErrorAs(t, Put(context.Background(), e, chunkObject1, false), &alreadyRemoved)
+}

pkg/local_object_storage/engine/put.go

@@ -71,21 +71,21 @@ func (e *StorageEngine) put(ctx context.Context, prm PutPrm) error {
 
 	// In #1146 this check was parallelized, however, it became
 	// much slower on fast machines for 4 shards.
-	var parent oid.Address
+	var ecParent oid.Address
 	if prm.Object.ECHeader() != nil {
-		parent.SetObject(prm.Object.ECHeader().Parent())
-		parent.SetContainer(addr.Container())
+		ecParent.SetObject(prm.Object.ECHeader().Parent())
+		ecParent.SetContainer(addr.Container())
 	}
 	var shPrm shard.ExistsPrm
 	shPrm.Address = addr
-	shPrm.ParentAddress = parent
+	shPrm.ECParentAddress = ecParent
 	existed, locked, err := e.exists(ctx, shPrm)
 	if err != nil {
 		return err
 	}
 
 	if !existed && locked {
-		lockers, err := e.GetLocked(ctx, parent)
+		lockers, err := e.GetLocked(ctx, ecParent)
 		if err != nil {
 			return err
 		}

pkg/local_object_storage/metabase/exists.go

@@ -20,8 +20,8 @@ import (
 
 // ExistsPrm groups the parameters of Exists operation.
 type ExistsPrm struct {
-	addr  oid.Address
-	paddr oid.Address
+	addr         oid.Address
+	ecParentAddr oid.Address
 }
 
 // ExistsRes groups the resulting values of Exists operation.
@@ -37,9 +37,9 @@ func (p *ExistsPrm) SetAddress(addr oid.Address) {
 	p.addr = addr
 }
 
-// SetParent is an Exists option to set objects parent.
-func (p *ExistsPrm) SetParent(addr oid.Address) {
-	p.paddr = addr
+// SetECParent is an Exists option to set objects parent.
+func (p *ExistsPrm) SetECParent(addr oid.Address) {
+	p.ecParentAddr = addr
 }
 
 // Exists returns the fact that the object is in the metabase.
@@ -82,7 +82,7 @@ func (db *DB) Exists(ctx context.Context, prm ExistsPrm) (res ExistsRes, err err
 	currEpoch := db.epochState.CurrentEpoch()
 
 	err = db.boltDB.View(func(tx *bbolt.Tx) error {
-		res.exists, res.locked, err = db.exists(tx, prm.addr, prm.paddr, currEpoch)
+		res.exists, res.locked, err = db.exists(tx, prm.addr, prm.ecParentAddr, currEpoch)
 
 		return err
 	})
@@ -90,10 +90,21 @@ func (db *DB) Exists(ctx context.Context, prm ExistsPrm) (res ExistsRes, err err
 	return res, metaerr.Wrap(err)
 }
 
-func (db *DB) exists(tx *bbolt.Tx, addr oid.Address, parent oid.Address, currEpoch uint64) (bool, bool, error) {
+func (db *DB) exists(tx *bbolt.Tx, addr oid.Address, ecParent oid.Address, currEpoch uint64) (bool, bool, error) {
 	var locked bool
-	if !parent.Equals(oid.Address{}) {
-		locked = objectLocked(tx, parent.Container(), parent.Object())
+	if !ecParent.Equals(oid.Address{}) {
+		st, err := objectStatus(tx, ecParent, currEpoch)
+		if err != nil {
+			return false, false, err
+		}
+		switch st {
+		case 2:
+			return false, locked, logicerr.Wrap(new(apistatus.ObjectAlreadyRemoved))
+		case 3:
+			return false, locked, ErrObjectIsExpired
+		}
+
+		locked = objectLocked(tx, ecParent.Container(), ecParent.Object())
 	}
 	// check graveyard and object expiration first
 	st, err := objectStatus(tx, addr, currEpoch)

pkg/local_object_storage/metabase/put.go

@@ -121,9 +121,15 @@ func (db *DB) put(tx *bbolt.Tx,
 		return PutRes{}, errors.New("missing container in object")
 	}
 
+	var ecParentAddress oid.Address
+	if ecHeader := obj.ECHeader(); ecHeader != nil {
+		ecParentAddress.SetContainer(cnr)
+		ecParentAddress.SetObject(ecHeader.Parent())
+	}
+
 	isParent := si != nil
 
-	exists, _, err := db.exists(tx, objectCore.AddressOf(obj), oid.Address{}, currEpoch)
+	exists, _, err := db.exists(tx, objectCore.AddressOf(obj), ecParentAddress, currEpoch)
 
 	var splitInfoError *objectSDK.SplitInfoError
 	if errors.As(err, &splitInfoError) {

pkg/local_object_storage/shard/exists.go

@@ -18,7 +18,7 @@ type ExistsPrm struct {
 	// Exists option to set object checked for existence.
 	Address oid.Address
 	// Exists option to set parent object checked for existence.
-	ParentAddress oid.Address
+	ECParentAddress oid.Address
 }
 
 // ExistsRes groups the resulting values of Exists operation.
@@ -74,7 +74,7 @@ func (s *Shard) Exists(ctx context.Context, prm ExistsPrm) (ExistsRes, error) {
 	} else {
 		var existsPrm meta.ExistsPrm
 		existsPrm.SetAddress(prm.Address)
-		existsPrm.SetParent(prm.ParentAddress)
+		existsPrm.SetECParent(prm.ECParentAddress)
 
 		var res meta.ExistsRes
 		res, err = s.metaBase.Exists(ctx, existsPrm)

pkg/services/common/ape/checker.go

@@ -11,7 +11,6 @@ import (
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/netmap"
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/ape"
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer"
-	apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
 	cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user"
 	apechain "git.frostfs.info/TrueCloudLab/policy-engine/pkg/chain"
@@ -104,14 +103,7 @@ func (c *checkerCoreImpl) CheckAPE(prm CheckPrm) error {
 	if found && status == apechain.Allow {
 		return nil
 	}
-	err = fmt.Errorf("access to operation %s is denied by access policy engine: %s", prm.Request.Operation(), status.String())
-	return apeErr(err)
-}
-
-func apeErr(err error) error {
-	errAccessDenied := &apistatus.ObjectAccessDenied{}
-	errAccessDenied.WriteReason(err.Error())
-	return errAccessDenied
+	return newChainRouterError(prm.Request.Operation(), status)
 }
 
 // isValidBearer checks whether bearer token was correctly signed by authorized

pkg/services/common/ape/error.go

@@ -0,0 +1,33 @@
+package ape
+
+import (
+	"fmt"
+
+	apechain "git.frostfs.info/TrueCloudLab/policy-engine/pkg/chain"
+)
+
+// ChainRouterError is returned when chain router validation prevents
+// the APE request from being processed (no rule found, access denied, etc.).
+type ChainRouterError struct {
+	operation string
+	status    apechain.Status
+}
+
+func (e *ChainRouterError) Error() string {
+	return fmt.Sprintf("access to operation %s is denied by access policy engine: %s", e.Operation(), e.Status())
+}
+
+func (e *ChainRouterError) Operation() string {
+	return e.operation
+}
+
+func (e *ChainRouterError) Status() apechain.Status {
+	return e.status
+}
+
+func newChainRouterError(operation string, status apechain.Status) *ChainRouterError {
+	return &ChainRouterError{
+		operation: operation,
+		status:    status,
+	}
+}

pkg/services/object/ape/errors.go

@@ -1,10 +1,19 @@
 package ape
 
 import (
+	"errors"
+
+	checkercore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/common/ape"
 	apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
 )
 
 func toStatusErr(err error) error {
+	var chRouterErr *checkercore.ChainRouterError
+	if !errors.As(err, &chRouterErr) {
+		errServerInternal := &apistatus.ServerInternal{}
+		apistatus.WriteInternalServerErr(errServerInternal, err)
+		return errServerInternal
+	}
 	errAccessDenied := &apistatus.ObjectAccessDenied{}
 	errAccessDenied.WriteReason("ape denied request: " + err.Error())
 	return errAccessDenied

pkg/services/object_manager/placement/cache_test.go

@@ -85,7 +85,10 @@ func TestContainerNodesCache(t *testing.T) {
 	})
 	t.Run("the error is propagated", func(t *testing.T) {
 		var pp netmapSDK.PlacementPolicy
-		require.NoError(t, pp.DecodeString("REP 1 SELECT 1 FROM X FILTER ATTR EQ 42 AS X"))
+		r := netmapSDK.ReplicaDescriptor{}
+		r.SetNumberOfObjects(1)
+		r.SetSelectorName("Missing")
+		pp.AddReplicas(r)
 
 		c := placement.NewContainerNodesCache(size)
 		_, err := c.ContainerNodes(nm(1, nodes[0:1]), cidtest.ID(), pp)

pkg/services/object_manager/placement/traverser.go

@@ -202,7 +202,7 @@ func sortVector(cfg *cfg, unsortedVector []netmap.NodeInfo) ([]netmap.NodeInfo,
 			metrics: m,
 		}
 	}
-	slices.SortFunc(nm, func(a, b nodeMetrics) int {
+	slices.SortStableFunc(nm, func(a, b nodeMetrics) int {
 		return slices.Compare(a.metrics, b.metrics)
 	})
 	sortedVector := make([]netmap.NodeInfo, len(unsortedVector))

pkg/services/policer/ec.go

@@ -281,6 +281,8 @@ func (p *Policer) adjustECPlacement(ctx context.Context, objInfo objectcore.Info
 				}
 				chunkIDs[ch.Index] = ecInfoChunkID
 			}
+		} else if client.IsErrObjectAlreadyRemoved(err) {
+			restore = false
 		} else if !p.cfg.netmapKeys.IsLocalKey(n.PublicKey()) && uint32(idx) < objInfo.ECInfo.Total {
 			p.log.Warn(ctx, logs.PolicerCouldNotGetObjectFromNodeMoving, zap.String("node", hex.EncodeToString(n.PublicKey())), zap.Stringer("object", parentAddress), zap.Error(err))
 			p.replicator.HandleReplicationTask(ctx, replicator.Task{

pkg/services/tree/signature.go

@@ -9,8 +9,10 @@ import (
 	"fmt"
 
 	core "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/container"
+	checkercore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/common/ape"
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/api/refs"
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer"
+	apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/acl"
 	cidSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
 	frostfscrypto "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/crypto"
@@ -62,7 +64,22 @@ func (s *Service) verifyClient(ctx context.Context, req message, cid cidSDK.ID,
 		return fmt.Errorf("can't get request role: %w", err)
 	}
 
-	return s.checkAPE(ctx, bt, cnr, cid, op, role, pubKey)
+	if err = s.checkAPE(ctx, bt, cnr, cid, op, role, pubKey); err != nil {
+		return apeErr(err)
+	}
+	return nil
+}
+
+func apeErr(err error) error {
+	var chRouterErr *checkercore.ChainRouterError
+	if !errors.As(err, &chRouterErr) {
+		errServerInternal := &apistatus.ServerInternal{}
+		apistatus.WriteInternalServerErr(errServerInternal, err)
+		return errServerInternal
+	}
+	errAccessDenied := &apistatus.ObjectAccessDenied{}
+	errAccessDenied.WriteReason(err.Error())
+	return errAccessDenied
 }
 
 // Returns true iff the operation is read-only and request was signed