object: Introduce new fields for ECHeader type #223

Merged
fyrchik merged 1 commit from aarifullin/frostfs-sdk-go:feat/ecHeader_parentSplit into master 2024-09-04 19:51:16 +00:00
4 changed files with 152 additions and 13 deletions

View file

@ -10,21 +10,25 @@ import (
// ECHeader represents erasure coding header. // ECHeader represents erasure coding header.
type ECHeader struct { 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. // 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{ 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) { func (e *ECHeader) WriteToV2(h *object.ECHeader) {
var parent refs.ObjectID var parent refs.ObjectID
e.parent.WriteToV2(&parent) 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.Parent = &parent
h.Index = e.index h.Index = e.index
h.Total = e.total h.Total = e.total
@ -49,6 +61,13 @@ func (e *ECHeader) ReadFromV2(h *object.ECHeader) error {
} }
_ = e.parent.ReadFromV2(*h.Parent) _ = 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.index = h.Index
e.total = h.Total e.total = h.Total
e.header = h.Header e.header = h.Header
@ -88,6 +107,22 @@ func (e *ECHeader) SetParent(id oid.ID) {
e.parent = 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 { func (e *ECHeader) Index() uint32 {
return e.index return e.index
} }

View file

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

View file

@ -4,6 +4,7 @@ import (
"crypto/ecdsa" "crypto/ecdsa"
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object" 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. // 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.SetPayload(payloadShards[i])
chunk.SetPayloadSize(uint64(len(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) chunk.SetECHeader(ec)
if err := setIDWithSignature(chunk, key); err != nil { if err := setIDWithSignature(chunk, key); err != nil {
return nil, err return nil, err

View file

@ -5,6 +5,7 @@ import (
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object" objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/erasurecode" "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/nspcc-dev/neo-go/pkg/crypto/keys"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
@ -41,5 +42,97 @@ func TestSplitMaxShardCount(t *testing.T) {
require.NoError(t, objectSDK.CheckHeaderVerificationFields(part)) 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())
}
})
} }