EC Policer (#3557)

Author: cthulhu-rider

cmd/neofs-lens/internal/storage/list.go

@@ -34,7 +34,7 @@ func listFunc(cmd *cobra.Command, _ []string) error {
 	defer storage.Close()
 
 	var (
-		addrs  []objectcore.AddressWithType
+		addrs  []objectcore.AddressWithAttributes
 		cursor *engine.Cursor
 	)
 	for {

cmd/neofs-lens/internal/storage/sanity.go

@@ -142,7 +142,7 @@ func sanityCheck(cmd *cobra.Command, _ []string) error {
 
 func checkShard(cmd *cobra.Command, sh storageShard) (int, error) {
 	var (
-		addrs          []objectcore.AddressWithType
+		addrs          []objectcore.AddressWithAttributes
 		cursor         *meta.Cursor
 		err            error
 		objectsChecked int

cmd/neofs-node/object.go

@@ -204,7 +204,7 @@ func initObjectService(c *cfg) {
 		policer.WithMaxCapacity(c.appCfg.Policer.MaxWorkers),
 		policer.WithPool(c.cfgObject.pool.replication),
 		policer.WithNodeLoader(c),
-		policer.WithNetwork(noEC{c}),
+		policer.WithNetwork(c),
 		policer.WithReplicationCooldown(c.appCfg.Policer.ReplicationCooldown),
 		policer.WithObjectBatchSize(c.appCfg.Policer.ObjectBatchSize),
 	)
@@ -695,32 +695,13 @@ func (o objectSource) SearchOne(ctx context.Context, cnr cid.ID, filters objectS
 // IsLocalNodePublicKey implements [getsvc.NeoFSNetwork].
 func (c *cfg) IsLocalNodePublicKey(b []byte) bool { return c.IsLocalKey(b) }
 
-// temporary wrapper until [getsvc.NeoFSNetwork] and [policer.Network]
-// interfaces converge.
-type noEC struct {
-	*cfg
-}
-
-// GetNodesForObject reads storage policy of the referenced container from the
-// underlying container storage, reads network map at the specified epoch from
-// the underlying storage, applies the storage policy to it and returns sorted
-// lists of selected storage nodes along with the policy rules.
-//
-// Resulting slices must not be changed.
-//
-// GetNodesForObject implements [policer.Network].
-func (x noEC) GetNodesForObject(addr oid.Address) ([][]netmapsdk.NodeInfo, []uint, error) {
-	nodeLists, repRules, _, err := x.cfg.GetNodesForObject(addr)
-	return nodeLists[:len(repRules)], repRules, err
-}
-
 // GetNodesForObject reads storage policy of the referenced container from the
 // underlying container storage, reads network map at the specified epoch from
 // the underlying storage, applies the storage policy to it and returns sorted
 // lists of selected storage nodes along with the per-list numbers of primary
 // object holders. Resulting slices must not be changed.
 //
-// GetNodesForObject implements [getsvc.NeoFSNetwork].
+// GetNodesForObject implements [getsvc.NeoFSNetwork], [policer.Network].
 func (c *cfg) GetNodesForObject(addr oid.Address) ([][]netmapsdk.NodeInfo, []uint, []iec.Rule, error) {
 	nodeSets, repRules, ecRules, err := c.cfgObject.containerNodes.getNodesForObject(addr)
 	return nodeSets, repRules, ecRules, err

internal/ec/object_test.go

@@ -172,3 +172,54 @@ func TestFormObjectForECPart(t *testing.T) {
 		require.Equal(t, checksum.NewTillichZemor(tz.Sum(part)), phh)
 	})
 }
+
+func TestDecodePartInfoFromAttributes(t *testing.T) {
+	t.Run("missing", func(t *testing.T) {
+		pi, err := iec.DecodePartInfoFromAttributes("", "")
+		require.NoError(t, err)
+		require.EqualValues(t, -1, pi.RuleIndex)
+	})
+
+	t.Run("failure", func(t *testing.T) {
+		for _, tc := range []struct {
+			name      string
+			ruleIdx   string
+			partIdx   string
+			assertErr func(t *testing.T, err error)
+		}{
+			{name: "non-int rule index", ruleIdx: "not_an_int", partIdx: "34", assertErr: func(t *testing.T, err error) {
+				require.EqualError(t, err, `decode rule index: strconv.ParseUint: parsing "not_an_int": invalid syntax`)
+			}},
+			{name: "negative rule index", ruleIdx: "-12", partIdx: "34", assertErr: func(t *testing.T, err error) {
+				require.EqualError(t, err, `decode rule index: strconv.ParseUint: parsing "-12": invalid syntax`)
+			}},
+			{name: "rule index overflow", ruleIdx: "256", partIdx: "34", assertErr: func(t *testing.T, err error) {
+				require.EqualError(t, err, "rule index out of range")
+			}},
+			{name: "non-int part index", ruleIdx: "12", partIdx: "not_an_int", assertErr: func(t *testing.T, err error) {
+				require.EqualError(t, err, `decode part index: strconv.ParseUint: parsing "not_an_int": invalid syntax`)
+			}},
+			{name: "negative part index", ruleIdx: "12", partIdx: "-34", assertErr: func(t *testing.T, err error) {
+				require.EqualError(t, err, `decode part index: strconv.ParseUint: parsing "-34": invalid syntax`)
+			}},
+			{name: "part index overflow", ruleIdx: "12", partIdx: "256", assertErr: func(t *testing.T, err error) {
+				require.EqualError(t, err, "part index out of range")
+			}},
+			{name: "rule index without part index", ruleIdx: "12", partIdx: "", assertErr: func(t *testing.T, err error) {
+				require.EqualError(t, err, "rule index is set, part index is not")
+			}},
+			{name: "part index without rule index", ruleIdx: "", partIdx: "34", assertErr: func(t *testing.T, err error) {
+				require.EqualError(t, err, "part index is set, rule index is not")
+			}},
+		} {
+			t.Run(tc.name, func(t *testing.T) {
+				_, err := iec.DecodePartInfoFromAttributes(tc.ruleIdx, tc.partIdx)
+				tc.assertErr(t, err)
+			})
+		}
+	})
+
+	pi, err := iec.DecodePartInfoFromAttributes("12", "34")
+	require.NoError(t, err)
+	require.Equal(t, iec.PartInfo{RuleIndex: 12, Index: 34}, pi)
+}

internal/ec/objects.go

@@ -1,7 +1,9 @@
 package ec
 
 import (
+	"errors"
 	"fmt"
+	"strconv"
 
 	iobject "github.com/nspcc-dev/neofs-node/internal/object"
 	"github.com/nspcc-dev/neofs-sdk-go/checksum"
@@ -73,3 +75,50 @@ func FormObjectForECPart(signer neofscrypto.Signer, parent object.Object, part [
 
 	return obj, nil
 }
+
+// DecodePartInfoFromAttributes decodes EC part info from given object
+// attributes. It one of attributes is set, the other must be set too. If both
+// are missing, DecodePartInfoFromAttributes returns [PartInfo.RuleIndex] = -1
+// without error.
+func DecodePartInfoFromAttributes(ruleIdxAttr, partIdxAttr string) (PartInfo, error) {
+	// TODO: sync with object GET server
+	// TODO: sync with GetPartInfo, it does not check for numeric limits.
+	if ruleIdxAttr == "" {
+		if partIdxAttr != "" {
+			return PartInfo{}, errors.New("part index is set, rule index is not")
+		}
+		return PartInfo{RuleIndex: -1}, nil
+	}
+	if partIdxAttr == "" {
+		return PartInfo{}, errors.New("rule index is set, part index is not")
+	}
+
+	ruleIdx, err := decodeUint8StringToInt(ruleIdxAttr)
+	if err != nil {
+		if errors.Is(err, strconv.ErrRange) {
+			return PartInfo{}, errors.New("rule index out of range")
+		}
+		return PartInfo{}, fmt.Errorf("decode rule index: %w", err)
+	}
+	partIdx, err := decodeUint8StringToInt(partIdxAttr)
+	if err != nil {
+		if errors.Is(err, strconv.ErrRange) {
+			return PartInfo{}, errors.New("part index out of range")
+		}
+		return PartInfo{}, fmt.Errorf("decode part index: %w", err)
+	}
+
+	return PartInfo{
+		RuleIndex: ruleIdx,
+		Index:     partIdx,
+	}, nil
+}
+
+// returns [strconv.ErrRange] if value >= 256.
+func decodeUint8StringToInt(s string) (int, error) {
+	n, err := strconv.ParseUint(s, 10, 8)
+	if err != nil {
+		return 0, err
+	}
+	return int(n), nil
+}

internal/slices/index.go

@@ -15,3 +15,12 @@ func Indexes(n int) []int {
 	}
 	return s
 }
+
+// CollectIndex returns new slice of shallows copies of s elements with given indexes.
+func CollectIndex[E any, S []E](s S, idxs ...int) S {
+	newS := make(S, len(idxs))
+	for i, idx := range idxs {
+		newS[i] = s[idx]
+	}
+	return newS
+}

internal/slices/index_test.go

@@ -1,6 +1,7 @@
 package slices_test
 
 import (
+	"slices"
 	"testing"
 
 	islices "github.com/nspcc-dev/neofs-node/internal/slices"
@@ -22,3 +23,23 @@ func TestIndexCombos(t *testing.T) {
 		{2, 3},
 	})
 }
+
+func TestCollectIndex(t *testing.T) {
+	t.Run("nil", func(t *testing.T) {
+		require.Empty(t, islices.CollectIndex([]any(nil)))
+	})
+	t.Run("empty", func(t *testing.T) {
+		require.Empty(t, islices.CollectIndex([]any{}))
+	})
+
+	s := []string{"0", "1", "2", "3", "4", "5", "6", "7", "8", "9"}
+	sc := slices.Clone(s)
+
+	newS := islices.CollectIndex(s, 9, 7, 5, 3, 1)
+	require.Equal(t, []string{"9", "7", "5", "3", "1"}, newS)
+	require.Len(t, newS, 5)
+	require.EqualValues(t, 5, cap(newS))
+
+	newS[0] += "_"
+	require.Equal(t, sc, s)
+}

internal/slices/slices.go

@@ -56,3 +56,17 @@ func RepeatElement[E any, S []E](n int, e E) S {
 	}
 	return s
 }
+
+// MaxLen returns max length in s. Returns 0 if s is empty.
+func MaxLen(s []string) int {
+	if len(s) == 0 {
+		return 0
+	}
+	mx := len(s[0])
+	for i := 1; i < len(s); i++ {
+		if len(s[i]) > mx {
+			mx = len(s[i])
+		}
+	}
+	return mx
+}

internal/slices/slices_test.go

@@ -2,11 +2,13 @@ package slices_test
 
 import (
 	"errors"
+	"math/rand/v2"
 	"slices"
 	"strconv"
 	"testing"
 
 	islices "github.com/nspcc-dev/neofs-node/internal/slices"
+	"github.com/nspcc-dev/neofs-node/internal/testutil"
 	"github.com/stretchr/testify/require"
 )
 
@@ -93,3 +95,28 @@ func TestRepeatElements(t *testing.T) {
 		})
 	}
 }
+
+func TestMaxLen(t *testing.T) {
+	t.Run("nil", func(t *testing.T) {
+		require.Zero(t, islices.MaxLen(nil))
+	})
+	t.Run("empty", func(t *testing.T) {
+		require.Zero(t, islices.MaxLen([]string{}))
+	})
+
+	s := make([]string, 100)
+	for i := range s {
+		s[i] = string(testutil.RandByteSlice(i))
+	}
+	require.EqualValues(t, 99, islices.MaxLen(s))
+
+	slices.Reverse(s)
+	require.EqualValues(t, 99, islices.MaxLen(s))
+
+	rand.Shuffle(len(s), func(i, j int) { s[i], s[j] = s[j], s[i] })
+	require.EqualValues(t, 99, islices.MaxLen(s))
+
+	e := s[50]
+	s = islices.RepeatElement(len(s), e)
+	require.EqualValues(t, len(e), islices.MaxLen(s))
+}

internal/testutil/neofs.go

@@ -7,10 +7,18 @@ import (
 )
 
 // Nodes returns n [netmap.NodeInfo] elements with unique public keys.
+//
+// Each node has two network addresses with localhost IP. Ports are sequential
+// starting from 10000.
 func Nodes(n int) []netmap.NodeInfo {
+	const portsStart = 10_000
 	nodes := make([]netmap.NodeInfo, n)
 	for i := range nodes {
 		nodes[i].SetPublicKey([]byte("public_key_" + strconv.Itoa(i)))
+		nodes[i].SetNetworkEndpoints(
+			"localhost:"+strconv.Itoa(portsStart+2*i),
+			"localhost:"+strconv.Itoa(portsStart+2*i+1),
+		)
 	}
 	return nodes
 }