[#223] core: Refactor object format validator

Resolve funlen linter for FormatValidator.ValidateContent method.

Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
Dmitrii Stepanov 2023-04-06 13:02:37 +03:00
parent 02831d427b
commit 6016d78a45

View file

@ -202,112 +202,24 @@ func (i ContentMeta) Objects() []oid.ID {
}
// ValidateContent validates payload content according to the object type.
//
// nolint: funlen
func (v *FormatValidator) ValidateContent(o *object.Object) (ContentMeta, error) {
meta := ContentMeta{
typ: o.Type(),
}
switch o.Type() {
case object.TypeRegular:
// ignore regular objects, they do not need payload formatting
case object.TypeTombstone:
if len(o.Payload()) == 0 {
return ContentMeta{}, fmt.Errorf("(%T) empty payload in tombstone", v)
}
tombstone := object.NewTombstone()
if err := tombstone.Unmarshal(o.Payload()); err != nil {
return ContentMeta{}, fmt.Errorf("(%T) could not unmarshal tombstone content: %w", v, err)
}
// check if the tombstone has the same expiration in the body and the header
exp, err := expirationEpochAttribute(o)
if err != nil {
if err := v.fillAndValidateTombstoneMeta(o, &meta); err != nil {
return ContentMeta{}, err
}
if exp != tombstone.ExpirationEpoch() {
return ContentMeta{}, errTombstoneExpiration
}
// mark all objects from the tombstone body as removed in the storage engine
_, ok := o.ContainerID()
if !ok {
return ContentMeta{}, errors.New("missing container ID")
}
idList := tombstone.Members()
meta.objs = idList
case object.TypeStorageGroup:
if len(o.Payload()) == 0 {
return ContentMeta{}, fmt.Errorf("(%T) empty payload in SG", v)
}
var sg storagegroup.StorageGroup
if err := sg.Unmarshal(o.Payload()); err != nil {
return ContentMeta{}, fmt.Errorf("(%T) could not unmarshal SG content: %w", v, err)
}
mm := sg.Members()
meta.objs = mm
lenMM := len(mm)
if lenMM == 0 {
return ContentMeta{}, errEmptySGMembers
}
uniqueFilter := make(map[oid.ID]struct{}, lenMM)
for i := 0; i < lenMM; i++ {
if _, alreadySeen := uniqueFilter[mm[i]]; alreadySeen {
return ContentMeta{}, fmt.Errorf("storage group contains non-unique member: %s", mm[i])
}
uniqueFilter[mm[i]] = struct{}{}
if err := v.fillAndValidateStorageGroupMeta(o, &meta); err != nil {
return ContentMeta{}, err
}
case object.TypeLock:
if len(o.Payload()) == 0 {
return ContentMeta{}, errors.New("empty payload in lock")
if err := v.fillAndValidateLockMeta(o, &meta); err != nil {
return ContentMeta{}, err
}
_, ok := o.ContainerID()
if !ok {
return ContentMeta{}, errors.New("missing container")
}
_, ok = o.ID()
if !ok {
return ContentMeta{}, errors.New("missing ID")
}
// check that LOCK object has correct expiration epoch
lockExp, err := expirationEpochAttribute(o)
if err != nil {
return ContentMeta{}, fmt.Errorf("lock object expiration epoch: %w", err)
}
if currEpoch := v.netState.CurrentEpoch(); lockExp < currEpoch {
return ContentMeta{}, fmt.Errorf("lock object expiration: %d; current: %d", lockExp, currEpoch)
}
var lock object.Lock
err = lock.Unmarshal(o.Payload())
if err != nil {
return ContentMeta{}, fmt.Errorf("decode lock payload: %w", err)
}
num := lock.NumberOfMembers()
if num == 0 {
return ContentMeta{}, errors.New("missing locked members")
}
meta.objs = make([]oid.ID, num)
lock.ReadMembers(meta.objs)
default:
// ignore all other object types, they do not need payload formatting
}
@ -315,6 +227,104 @@ func (v *FormatValidator) ValidateContent(o *object.Object) (ContentMeta, error)
return meta, nil
}
func (v *FormatValidator) fillAndValidateLockMeta(o *object.Object, meta *ContentMeta) error {
if len(o.Payload()) == 0 {
return errors.New("empty payload in lock")
}
if _, ok := o.ContainerID(); !ok {
return errors.New("missing container")
}
if _, ok := o.ID(); !ok {
return errors.New("missing ID")
}
// check that LOCK object has correct expiration epoch
lockExp, err := expirationEpochAttribute(o)
if err != nil {
return fmt.Errorf("lock object expiration epoch: %w", err)
}
if currEpoch := v.netState.CurrentEpoch(); lockExp < currEpoch {
return fmt.Errorf("lock object expiration: %d; current: %d", lockExp, currEpoch)
}
var lock object.Lock
if err = lock.Unmarshal(o.Payload()); err != nil {
return fmt.Errorf("decode lock payload: %w", err)
}
num := lock.NumberOfMembers()
if num == 0 {
return errors.New("missing locked members")
}
meta.objs = make([]oid.ID, num)
lock.ReadMembers(meta.objs)
return nil
}
func (v *FormatValidator) fillAndValidateStorageGroupMeta(o *object.Object, meta *ContentMeta) error {
if len(o.Payload()) == 0 {
return fmt.Errorf("(%T) empty payload in storage group", v)
}
var sg storagegroup.StorageGroup
if err := sg.Unmarshal(o.Payload()); err != nil {
return fmt.Errorf("(%T) could not unmarshal storage group content: %w", v, err)
}
mm := sg.Members()
meta.objs = mm
lenMM := len(mm)
if lenMM == 0 {
return errEmptySGMembers
}
uniqueFilter := make(map[oid.ID]struct{}, lenMM)
for i := 0; i < lenMM; i++ {
if _, alreadySeen := uniqueFilter[mm[i]]; alreadySeen {
return fmt.Errorf("storage group contains non-unique member: %s", mm[i])
}
uniqueFilter[mm[i]] = struct{}{}
}
return nil
}
func (v *FormatValidator) fillAndValidateTombstoneMeta(o *object.Object, meta *ContentMeta) error {
if len(o.Payload()) == 0 {
return fmt.Errorf("(%T) empty payload in tombstone", v)
}
tombstone := object.NewTombstone()
if err := tombstone.Unmarshal(o.Payload()); err != nil {
return fmt.Errorf("(%T) could not unmarshal tombstone content: %w", v, err)
}
// check if the tombstone has the same expiration in the body and the header
exp, err := expirationEpochAttribute(o)
if err != nil {
return err
}
if exp != tombstone.ExpirationEpoch() {
return errTombstoneExpiration
}
// mark all objects from the tombstone body as removed in the storage engine
if _, ok := o.ContainerID(); !ok {
return errors.New("missing container ID")
}
meta.objs = tombstone.Members()
return nil
}
var errExpired = errors.New("object has expired")
func (v *FormatValidator) checkExpiration(obj *object.Object) error {