diff --git a/cmd/neofs-node/control.go b/cmd/neofs-node/control.go index 52330b3e8..5e6c76204 100644 --- a/cmd/neofs-node/control.go +++ b/cmd/neofs-node/control.go @@ -36,6 +36,7 @@ func initControlService(c *cfg) { controlSvc.WithDeletedObjectHandler(func(addrList []oid.Address) error { var prm engine.DeletePrm prm.WithAddresses(addrList...) + prm.WithForceRemoval() _, err := c.cfgObject.cfgLocalStorage.localStorage.Delete(prm) diff --git a/pkg/local_object_storage/engine/delete.go b/pkg/local_object_storage/engine/delete.go index def9d284a..5b1ccc370 100644 --- a/pkg/local_object_storage/engine/delete.go +++ b/pkg/local_object_storage/engine/delete.go @@ -11,6 +11,8 @@ import ( // DeletePrm groups the parameters of Delete operation. type DeletePrm struct { addr []oid.Address + + forceRemoval bool } // DeleteRes groups the resulting values of Delete operation. @@ -25,6 +27,15 @@ func (p *DeletePrm) WithAddresses(addr ...oid.Address) { } } +// WithForceRemoval is a Delete option to remove an object despite any +// restrictions imposed on deleting that object. Expected to be used +// only in control service. +func (p *DeletePrm) WithForceRemoval() { + if p != nil { + p.forceRemoval = true + } +} + // Delete marks the objects to be removed. // // Returns an error if executions are blocked (see BlockExecution). @@ -65,6 +76,9 @@ func (e *StorageEngine) delete(prm DeletePrm) (DeleteRes, error) { var shPrm shard.InhumePrm shPrm.MarkAsGarbage(prm.addr[i]) + if prm.forceRemoval { + shPrm.ForceRemoval() + } _, err = sh.Inhume(shPrm) if err != nil { diff --git a/pkg/local_object_storage/metabase/inhume.go b/pkg/local_object_storage/metabase/inhume.go index 4c370c065..2c20fdee0 100644 --- a/pkg/local_object_storage/metabase/inhume.go +++ b/pkg/local_object_storage/metabase/inhume.go @@ -18,6 +18,8 @@ type InhumePrm struct { target []oid.Address lockObjectHandling bool + + forceRemoval bool } // InhumeRes encapsulates results of Inhume operation. @@ -55,6 +57,7 @@ func (p *InhumePrm) WithTombstoneAddress(addr oid.Address) { func (p *InhumePrm) WithGCMark() { if p != nil { p.tomb = nil + p.forceRemoval = false } } @@ -66,6 +69,15 @@ func (p *InhumePrm) WithLockObjectHandling() { } } +// WithForceGCMark allows removal any object. Expected to be +// called only in control service. +func (p *InhumePrm) WithForceGCMark() { + if p != nil { + p.tomb = nil + p.forceRemoval = true + } +} + // Inhume inhumes the object by specified address. // // tomb should not be nil. @@ -130,6 +142,19 @@ func (db *DB) Inhume(prm InhumePrm) (res InhumeRes, err error) { return apistatus.ObjectLocked{} } + var lockWasChecked bool + + // prevent lock objects to be inhumed + // if `Inhume` was called not with the + // `WithForceGCMark` option + if !prm.forceRemoval { + if isLockObject(tx, cnr, id) { + return fmt.Errorf("lock object removal, CID: %s, OID: %s", cnr, id) + } + + lockWasChecked = true + } + obj, err := db.get(tx, prm.target[i], false, true) // if object is stored and it is regular object then update bucket @@ -184,6 +209,14 @@ func (db *DB) Inhume(prm InhumePrm) (res InhumeRes, err error) { } if prm.lockObjectHandling { + // do not perform lock check if + // it was already called + if lockWasChecked { + // inhumed object is not of + // the LOCK type + continue + } + if isLockObject(tx, cnr, id) { res.deletedLockObj = append(res.deletedLockObj, prm.target[i]) } diff --git a/pkg/local_object_storage/shard/inhume.go b/pkg/local_object_storage/shard/inhume.go index 505027c77..ef4c7bb20 100644 --- a/pkg/local_object_storage/shard/inhume.go +++ b/pkg/local_object_storage/shard/inhume.go @@ -11,8 +11,9 @@ import ( // InhumePrm encapsulates parameters for inhume operation. type InhumePrm struct { - target []oid.Address - tombstone *oid.Address + target []oid.Address + tombstone *oid.Address + forceRemoval bool } // InhumeRes encapsulates results of inhume operation. @@ -40,6 +41,15 @@ func (p *InhumePrm) MarkAsGarbage(addr ...oid.Address) { } } +// ForceRemoval forces object removing despite any restrictions imposed +// on deleting that object. Expected to be used only in control service. +func (p *InhumePrm) ForceRemoval() { + if p != nil { + p.tombstone = nil + p.forceRemoval = true + } +} + // Inhume calls metabase. Inhume method to mark object as removed. It won't be // removed physically from blobStor and metabase until `Delete` operation. // @@ -68,6 +78,10 @@ func (s *Shard) Inhume(prm InhumePrm) (InhumeRes, error) { metaPrm.WithGCMark() } + if prm.forceRemoval { + metaPrm.WithForceGCMark() + } + res, err := s.metaBase.Inhume(metaPrm) if err != nil { s.log.Debug("could not mark object to delete in metabase",