[#223] object: Introduce new fields for ECHeader

* Introduce `parentSplitID`, `parentSplitParentID` fields
  for `ECHeader`;
* Fix ECHeader's constructor;
* Fix `Split` and `Reconstruct`;
* Add unit-tests.

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

object/erasure_code.go

@@ -10,21 +10,25 @@ import (
 
 // ECHeader represents erasure coding header.
 type ECHeader struct {
-	parent       oid.ID
+	parent              oid.ID
-	index        uint32
+	parentSplitID       *SplitID
-	total        uint32
+	parentSplitParentID *oid.ID
-	header       []byte
+	index               uint32
-	headerLength uint32
+	total               uint32
+	header              []byte
+	headerLength        uint32
 }
 
 // NewECHeader constructs new erasure coding header.
-func NewECHeader(parent oid.ID, index, total uint32, header []byte, headerLength uint32) *ECHeader {
+func NewECHeader(parent oid.ID, parentSplitID *SplitID, parentSplitParentID *oid.ID, index, total uint32, header []byte, headerLength uint32) *ECHeader {
 	return &ECHeader{
-		parent:       parent,
+		parent:              parent,
-		index:        index,
+		parentSplitID:       parentSplitID,
-		total:        total,
+		parentSplitParentID: parentSplitParentID,
-		header:       header,
+		index:               index,
-		headerLength: headerLength,
+		total:               total,
+		header:              header,
+		headerLength:        headerLength,
 	}
 }
 
@@ -32,6 +36,14 @@ func NewECHeader(parent oid.ID, index, total uint32, header []byte, headerLength
 func (e *ECHeader) WriteToV2(h *object.ECHeader) {
 	var parent refs.ObjectID
 	e.parent.WriteToV2(&parent)
+	h.ParentSplitID = e.parentSplitID.ToV2()
+
+	if e.parentSplitParentID != nil {
+		parentSplitParentID := new(refs.ObjectID)
+		e.parentSplitParentID.WriteToV2(parentSplitParentID)
+		h.ParentSplitParentID = parentSplitParentID
+	}
+
 	h.Parent = &parent
 	h.Index = e.index
 	h.Total = e.total
@@ -49,6 +61,13 @@ func (e *ECHeader) ReadFromV2(h *object.ECHeader) error {
 	}
 
 	_ = e.parent.ReadFromV2(*h.Parent)
+	e.parentSplitID = NewSplitIDFromV2(h.ParentSplitID)
+	if h.ParentSplitParentID != nil {
+		if e.parentSplitParentID == nil {
+			e.parentSplitParentID = new(oid.ID)
+		}
+		_ = e.parentSplitParentID.ReadFromV2(*h.ParentSplitParentID)
+	}
 	e.index = h.Index
 	e.total = h.Total
 	e.header = h.Header
@@ -88,6 +107,22 @@ func (e *ECHeader) SetParent(id oid.ID) {
 	e.parent = id
 }
 
+func (e *ECHeader) ParentSplitID() *SplitID {
+	return e.parentSplitID
+}
+
+func (e *ECHeader) SetParentSplitID(parentSplitID *SplitID) {
+	e.parentSplitID = parentSplitID
+}
+
+func (e *ECHeader) ParentSplitParentID() *oid.ID {
+	return e.parentSplitParentID
+}
+
+func (e *ECHeader) SetParentSplitParentID(parentSplitParentID *oid.ID) {
+	e.parentSplitParentID = parentSplitParentID
+}
+
 func (e *ECHeader) Index() uint32 {
 	return e.index
 }

object/erasurecode/reconstruct.go

@@ -87,6 +87,8 @@ func (c *Constructor) ReconstructParts(parts []*objectSDK.Object, required []boo
 	ec := parts[nonNilPart].GetECHeader()
 	parent := ec.Parent()
 	total := ec.Total()
+	splitID := ec.ParentSplitID()
+	parSplitParID := ec.ParentSplitParentID()
 
 	for i := range required {
 		if parts[i] != nil || !required[i] {
@@ -97,7 +99,7 @@ func (c *Constructor) ReconstructParts(parts []*objectSDK.Object, required []boo
 		copyRequiredFields(part, parts[nonNilPart])
 		part.SetPayload(c.payloadShards[i])
 		part.SetPayloadSize(uint64(len(c.payloadShards[i])))
-		part.SetECHeader(objectSDK.NewECHeader(parent, uint32(i), total,
+		part.SetECHeader(objectSDK.NewECHeader(parent, splitID, parSplitParID, uint32(i), total,
 			c.headerShards[i], c.headerLength))
 
 		if err := setIDWithSignature(part, key); err != nil {

object/erasurecode/split.go

@@ -4,6 +4,7 @@ import (
 	"crypto/ecdsa"
 
 	objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
+	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
 )
 
 // Split splits fully formed object into multiple chunks.
@@ -32,7 +33,15 @@ func (c *Constructor) Split(obj *objectSDK.Object, key *ecdsa.PrivateKey) ([]*ob
 		chunk.SetPayload(payloadShards[i])
 		chunk.SetPayloadSize(uint64(len(payloadShards[i])))
 
-		ec := objectSDK.NewECHeader(parent, uint32(i), uint32(len(payloadShards)), headerShards[i], uint32(len(header)))
+		var parentSplitParentID *oid.ID
+		if obj.HasParent() {
+			splitParentID, ok := obj.Parent().ID()
+			if ok {
+				parentSplitParentID = &splitParentID
+			}
+		}
+
+		ec := objectSDK.NewECHeader(parent, obj.SplitID(), parentSplitParentID, uint32(i), uint32(len(payloadShards)), headerShards[i], uint32(len(header)))
 		chunk.SetECHeader(ec)
 		if err := setIDWithSignature(chunk, key); err != nil {
 			return nil, err

object/erasurecode/split_test.go

@@ -5,6 +5,7 @@ import (
 
 	objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
 	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/erasurecode"
+	oidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id/test"
 	"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
 	"github.com/stretchr/testify/require"
 )
@@ -41,5 +42,97 @@ func TestSplitMaxShardCount(t *testing.T) {
 			require.NoError(t, objectSDK.CheckHeaderVerificationFields(part))
 		}
 	})
+	t.Run("ec parents are children of last Split part", func(t *testing.T) {
+		c, err := erasurecode.NewConstructor(1, erasurecode.MaxShardCount-1)
+		require.NoError(t, err)
+
+		splitted1 := newObject(t, 1024, pk)
+		splitted2 := newObject(t, 1024, pk)
+
+		splittedId1, _ := splitted1.ID()
+		splittedId2, _ := splitted2.ID()
+
+		splitID := objectSDK.NewSplitID()
+		parent := objectSDK.New()
+		parent.SetID(oidtest.ID())
+		parent.SetChildren(splittedId1, splittedId2)
+		parent.SetSplitID(splitID)
+
+		splitted1.SetSplitID(splitID)
+		splitted1.SetParent(parent)
+		splitted2.SetSplitID(splitID)
+		splitted2.SetParent(parent)
+
+		parts1, err := c.Split(splitted1, &pk.PrivateKey)
+		require.NoError(t, err)
+		require.Len(t, parts1, erasurecode.MaxShardCount)
+
+		for _, part := range parts1 {
+			require.NoError(t, objectSDK.CheckHeaderVerificationFields(part))
+
+			require.NotNil(t, part.ECHeader().ParentSplitID())
+			require.Equal(t, *splitID, *part.ECHeader().ParentSplitID())
+			require.NotNil(t, part.ECHeader().ParentSplitParentID())
+		}
+
+		parts2, err := c.Split(splitted2, &pk.PrivateKey)
+		require.NoError(t, err)
+		require.Len(t, parts1, erasurecode.MaxShardCount)
+
+		for _, part := range parts2 {
+			require.NoError(t, objectSDK.CheckHeaderVerificationFields(part))
+
+			require.NotNil(t, part.ECHeader().ParentSplitID())
+			require.Equal(t, *splitID, *part.ECHeader().ParentSplitID())
+			require.NotNil(t, part.ECHeader().ParentSplitParentID())
+		}
+
+	})
+	t.Run("ec parents are children of non-last Split part", func(t *testing.T) {
+		c, err := erasurecode.NewConstructor(1, erasurecode.MaxShardCount-1)
+		require.NoError(t, err)
+
+		splitted1 := newObject(t, 1024, pk)
+		splitted2 := newObject(t, 1024, pk)
+
+		splittedId1, _ := splitted1.ID()
+		splittedId2, _ := splitted2.ID()
+
+		splitID := objectSDK.NewSplitID()
+		parent := objectSDK.New()
+		// Such parent has got no ID.
+		parent.SetChildren(splittedId1, splittedId2)
+		parent.SetSplitID(splitID)
+
+		splitted1.SetSplitID(splitID)
+		splitted1.SetParent(parent)
+		splitted2.SetSplitID(splitID)
+		splitted2.SetParent(parent)
+
+		parts1, err := c.Split(splitted1, &pk.PrivateKey)
+		require.NoError(t, err)
+		require.Len(t, parts1, erasurecode.MaxShardCount)
+
+		for _, part := range parts1 {
+			require.NoError(t, objectSDK.CheckHeaderVerificationFields(part))
+
+			require.NotNil(t, part.ECHeader().ParentSplitID())
+			require.Equal(t, *splitID, *part.ECHeader().ParentSplitID())
+			require.Nil(t, part.ECHeader().ParentSplitParentID())
+		}
+
+		parts2, err := c.Split(splitted2, &pk.PrivateKey)
+		require.NoError(t, err)
+		require.Len(t, parts1, erasurecode.MaxShardCount)
+
+		for _, part := range parts2 {
+			require.NoError(t, objectSDK.CheckHeaderVerificationFields(part))
+
+			require.NotNil(t, part.ECHeader().ParentSplitID())
+			require.Equal(t, *splitID, *part.ECHeader().ParentSplitID())
+			require.Nil(t, part.ECHeader().ParentSplitParentID())
+		}
+
+	})
 
 }