sn/policer: Recreate unavailable EC parts in container when possible

Extend `Policer` with workers checking availability of EC parts. Free
worker takes the job each time some EC part is processed by regular
Policer routine. If no more than N parts are unavailable (where N is a
number of parity parts in the container policy), they are recreated
using available ones. Recreated object has the same ID but signed by SN
itself. To ensure such objects are accepted, the corresponding checks
are relaxed for EC parts.

Closes #3554.

Signed-off-by: Leonard Lyubich <[email protected]>

Author: cthulhu-rider

cmd/neofs-node/object.go

@@ -199,7 +199,7 @@ func initObjectService(c *cfg) {
 		),
 	)
 
-	c.policer = policer.New(
+	c.policer = policer.New(neofsecdsa.Signer(c.key.PrivateKey),
 		policer.WithLogger(c.log),
 		policer.WithLocalStorage(ls),
 		policer.WithRemoteHeader(

internal/crypto/object.go

@@ -13,7 +13,7 @@ import (
 )
 
 // AuthenticateObject checks whether obj is signed correctly by its owner.
-func AuthenticateObject(obj object.Object, fsChain HistoricN3ScriptRunner) error {
+func AuthenticateObject(obj object.Object, fsChain HistoricN3ScriptRunner, ecPart bool) error {
 	sig := obj.Signature()
 	if sig == nil {
 		return errMissingSignature
@@ -39,7 +39,7 @@ func AuthenticateObject(obj object.Object, fsChain HistoricN3ScriptRunner) error
 
 	if sessionToken != nil {
 		// NOTE: update this place for non-ECDSA schemes
-		if !sessionToken.AssertAuthKey((*neofsecdsa.PublicKey)(ecdsaPub)) { // same format for all ECDSA schemes
+		if !ecPart && !sessionToken.AssertAuthKey((*neofsecdsa.PublicKey)(ecdsaPub)) { // same format for all ECDSA schemes
 			return errors.New("session token is not for object's signer")
 		}
 		if err := AuthenticateToken(sessionToken, fsChain); err != nil {
@@ -57,7 +57,7 @@ func AuthenticateObject(obj object.Object, fsChain HistoricN3ScriptRunner) error
 		if !verifyECDSAFns[scheme](*ecdsaPub, sig.Value(), obj.GetID().Marshal()) {
 			return schemeError(scheme, errSignatureMismatch)
 		}
-		if sessionToken == nil && user.NewFromECDSAPublicKey(*ecdsaPub) != obj.Owner() {
+		if sessionToken == nil && !ecPart && user.NewFromECDSAPublicKey(*ecdsaPub) != obj.Owner() {
 			return errors.New("owner mismatches signature")
 		}
 	case neofscrypto.N3:

internal/crypto/object_test.go

@@ -20,14 +20,14 @@ import (
 func TestAuthenticateObject(t *testing.T) {
 	t.Run("without signature", func(t *testing.T) {
 		obj := getUnsignedObject()
-		require.EqualError(t, icrypto.AuthenticateObject(obj, nil), "missing signature")
+		require.EqualError(t, icrypto.AuthenticateObject(obj, nil, false), "missing signature")
 	})
 	t.Run("unsupported scheme", func(t *testing.T) {
 		obj := objectECDSASHA512
 		sig := *obj.Signature()
 		sig.SetScheme(4)
 		obj.SetSignature(&sig)
-		require.EqualError(t, icrypto.AuthenticateObject(obj, nil), "unsupported scheme 4")
+		require.EqualError(t, icrypto.AuthenticateObject(obj, nil, false), "unsupported scheme 4")
 	})
 	t.Run("invalid public key", func(t *testing.T) {
 		for _, tc := range []struct {
@@ -48,7 +48,7 @@ func TestAuthenticateObject(t *testing.T) {
 				sig := *obj.Signature()
 				sig.SetPublicKeyBytes(tc.changePub(sig.PublicKeyBytes()))
 				obj.SetSignature(&sig)
-				err := icrypto.AuthenticateObject(obj, nil)
+				err := icrypto.AuthenticateObject(obj, nil, false)
 				require.EqualError(t, err, "scheme ECDSA_SHA512: decode public key: "+tc.err)
 			})
 		}
@@ -70,7 +70,7 @@ func TestAuthenticateObject(t *testing.T) {
 					cp[i]++
 					sig.SetValue(cp)
 					tc.obj.SetSignature(&sig)
-					err := icrypto.AuthenticateObject(tc.obj, nil)
+					err := icrypto.AuthenticateObject(tc.obj, nil, false)
 					require.EqualError(t, err, fmt.Sprintf("scheme %s: signature mismatch", tc.scheme))
 				}
 			})
@@ -86,7 +86,7 @@ func TestAuthenticateObject(t *testing.T) {
 			{scheme: neofscrypto.ECDSA_WALLETCONNECT, object: wrongOwnerObjectECDSAWalletConnect},
 		} {
 			t.Run(tc.scheme.String(), func(t *testing.T) {
-				require.EqualError(t, icrypto.AuthenticateObject(tc.object, nil), "owner mismatches signature")
+				require.EqualError(t, icrypto.AuthenticateObject(tc.object, nil, false), "owner mismatches signature")
 			})
 		}
 	})
@@ -101,7 +101,7 @@ func TestAuthenticateObject(t *testing.T) {
 				{scheme: neofscrypto.ECDSA_WALLETCONNECT, object: objectWithNoIssuerSessionECDSAWalletConnect},
 			} {
 				t.Run(tc.scheme.String(), func(t *testing.T) {
-					require.EqualError(t, icrypto.AuthenticateObject(tc.object, nil), "session token: missing issuer")
+					require.EqualError(t, icrypto.AuthenticateObject(tc.object, nil, false), "session token: missing issuer")
 				})
 			}
 		})
@@ -115,7 +115,7 @@ func TestAuthenticateObject(t *testing.T) {
 				{scheme: neofscrypto.ECDSA_WALLETCONNECT, object: objectWithWrongIssuerSessionECDSAWalletConnect},
 			} {
 				t.Run(tc.scheme.String(), func(t *testing.T) {
-					require.EqualError(t, icrypto.AuthenticateObject(tc.object, nil), "session token: issuer mismatches signature")
+					require.EqualError(t, icrypto.AuthenticateObject(tc.object, nil, false), "session token: issuer mismatches signature")
 				})
 			}
 		})
@@ -129,7 +129,7 @@ func TestAuthenticateObject(t *testing.T) {
 				{scheme: neofscrypto.ECDSA_WALLETCONNECT, object: objectWithWrongSessionSubjectECDSAWalletConnect},
 			} {
 				t.Run(tc.scheme.String(), func(t *testing.T) {
-					require.EqualError(t, icrypto.AuthenticateObject(tc.object, nil), "session token is not for object's signer")
+					require.EqualError(t, icrypto.AuthenticateObject(tc.object, nil, false), "session token is not for object's signer")
 				})
 			}
 		})
@@ -143,7 +143,7 @@ func TestAuthenticateObject(t *testing.T) {
 				{scheme: neofscrypto.ECDSA_WALLETCONNECT, object: objectWithWrongOwnerSessionECDSAWalletConnect},
 			} {
 				t.Run(tc.scheme.String(), func(t *testing.T) {
-					require.EqualError(t, icrypto.AuthenticateObject(tc.object, nil), "different object owner and session issuer")
+					require.EqualError(t, icrypto.AuthenticateObject(tc.object, nil, false), "different object owner and session issuer")
 				})
 			}
 		})
@@ -160,7 +160,7 @@ func TestAuthenticateObject(t *testing.T) {
 		{name: neofscrypto.ECDSA_WALLETCONNECT.String() + " with session", object: objectWithSessionECDSAWalletConnect},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
-			require.NoError(t, icrypto.AuthenticateObject(tc.object, nil))
+			require.NoError(t, icrypto.AuthenticateObject(tc.object, nil, false))
 		})
 	}
 }

internal/ec/ec.go

@@ -112,6 +112,25 @@ func DecodeRange(rule Rule, fromIdx, toIdx int, parts [][]byte) error {
 	return nil
 }
 
+// DecodeIndexes decodes specified EC parts obtained by applying specified rule.
+func DecodeIndexes(rule Rule, parts [][]byte, idxs []int) error {
+	rs, err := newCoderForRule(rule)
+	if err != nil {
+		return err
+	}
+
+	required := make([]bool, rule.DataPartNum+rule.ParityPartNum)
+	for i := range idxs {
+		required[idxs[i]] = true
+	}
+
+	if err := rs.ReconstructSome(parts, required); err != nil {
+		return fmt.Errorf("restore Reed-Solomon: %w", err)
+	}
+
+	return nil
+}
+
 func newCoderForRule(rule Rule) (reedsolomon.Encoder, error) {
 	enc, err := reedsolomon.New(int(rule.DataPartNum), int(rule.ParityPartNum))
 	if err != nil { // should never happen with correct rule

pkg/core/object/ec_test.go

@@ -249,6 +249,19 @@ func TestFormatValidator_Validate_EC(t *testing.T) {
 		require.EqualError(t, v.Validate(&cp, false), "object with EC attributes __NEOFS__EC_RULE_IDX in container without EC rules")
 	})
 
+	t.Run("part created by 3rd party", func(t *testing.T) {
+		otherCreator := usertest.User()
+		for i := range parts {
+			obj, err := iec.FormObjectForECPart(otherCreator, parent, parts[i], iec.PartInfo{
+				RuleIndex: ruleIdx,
+				Index:     i,
+			})
+			require.NoError(t, err)
+
+			require.NoError(t, v.Validate(&obj, false))
+		}
+	})
+
 	for i := range ecParts {
 		require.NoError(t, v.Validate(&ecParts[i], false))
 	}

pkg/core/object/fmt.go

@@ -235,7 +235,7 @@ func (v *FormatValidator) validate(obj *object.Object, unprepared, isParent bool
 		if err := icrypto.AuthenticateObject(*obj, historicN3ScriptRunner{
 			FSChain:        v.fsChain,
 			NetmapContract: v.netmapContract,
-		}); err != nil {
+		}, isEC); err != nil {
 			return fmt.Errorf("authenticate: %w", err)
 		}
 

pkg/local_object_storage/engine/ec.go

@@ -203,3 +203,75 @@ loop:
 
 	return 0, nil, apistatus.ErrObjectNotFound
 }
+
+// HeadECPart is similar to [StorageEngine.GetECPart] but returns only the header.
+func (e *StorageEngine) HeadECPart(cnr cid.ID, parent oid.ID, pi iec.PartInfo) (object.Object, error) {
+	if e.metrics != nil {
+		defer elapsed(e.metrics.AddHeadECPartDuration)()
+	}
+
+	e.blockMtx.RLock()
+	defer e.blockMtx.RUnlock()
+	if e.blockErr != nil {
+		return object.Object{}, e.blockErr
+	}
+
+	// TODO: sync placement with PUT. They should sort shards equally, but now PUT sorts by part ID.
+	//  https://github.com/nspcc-dev/neofs-node/issues/3537
+	s := e.sortShardsFn(e, oid.NewAddress(cnr, parent))
+
+	var partID oid.ID
+loop:
+	for i := range s {
+		hdr, err := s[i].shardIface.HeadECPart(cnr, parent, pi)
+		switch {
+		case err == nil:
+			return hdr, nil
+		case errors.Is(err, apistatus.ErrObjectAlreadyRemoved):
+			return object.Object{}, err
+		case errors.Is(err, meta.ErrObjectIsExpired):
+			return object.Object{}, apistatus.ErrObjectNotFound // like Get
+		case errors.As(err, (*ierrors.ObjectID)(&partID)):
+			if partID.IsZero() {
+				panic("zero object ID returned as error")
+			}
+
+			e.log.Info("EC part's object ID resolved in shard but reading failed, continue by ID",
+				zap.Stringer("container", cnr), zap.Stringer("parent", parent),
+				zap.Int("ecRule", pi.RuleIndex), zap.Int("partIdx", pi.Index),
+				zap.Stringer("shardID", s[i].shardIface.ID()), zap.Error(err))
+			// TODO: need report error? Same for other places. https://github.com/nspcc-dev/neofs-node/issues/3538
+
+			s = s[i+1:]
+			break loop
+		case errors.Is(err, apistatus.ErrObjectNotFound):
+		default:
+			e.log.Info("failed to get EC part header from shard, ignore error",
+				zap.Stringer("container", cnr), zap.Stringer("parent", parent),
+				zap.Int("ecRule", pi.RuleIndex), zap.Int("partIdx", pi.Index),
+				zap.Stringer("shardID", s[i].shardIface.ID()), zap.Error(err))
+		}
+	}
+
+	if partID.IsZero() {
+		return object.Object{}, apistatus.ErrObjectNotFound
+	}
+
+	for i := range s {
+		// get an object bypassing the metabase. We can miss deletion or expiration mark. Get behaves like this, so here too.
+		hdr, err := s[i].shardIface.Head(oid.NewAddress(cnr, partID), true)
+		switch {
+		case err == nil:
+			return *hdr, nil
+		case errors.Is(err, apistatus.ErrObjectNotFound):
+		default:
+			e.log.Info("failed to get EC part header from shard bypassing metabase, ignore error",
+				zap.Stringer("container", cnr), zap.Stringer("parent", parent),
+				zap.Int("ecRule", pi.RuleIndex), zap.Int("partIdx", pi.Index),
+				zap.Stringer("partID", partID),
+				zap.Stringer("shardID", s[i].shardIface.ID()), zap.Error(err))
+		}
+	}
+
+	return object.Object{}, apistatus.ErrObjectNotFound
+}