Merge 48fbe5c into 2f5e83b

CHANGELOG.md

@@ -39,6 +39,7 @@ minor release, the component will be purged, so be prepared (see `Updating` sect
 - `neofs-lens write-cache list` command duplication (#2505)
 - `neofs-adm` works with contract wallet in `init` and `update-contracts` commands only (#2134)
 - Missing removed but locked objects in `SEARCH`'s results (#2526)
+- LOCK objects and regular objects expiration conflicts (#2392)
 
 ### Removed
 - Deprecated `morph.rpc_endpoint` SN and `morph.endpoint.client` IR config sections (#2400)

cmd/neofs-node/object.go

@@ -529,6 +529,10 @@ type engineWithNotifications struct {
  defaultTopic string
 }
 
+func (e engineWithNotifications) IsLocked(address oid.Address) (bool, error) {
+ return e.base.IsLocked(address)
+}
+
 func (e engineWithNotifications) Delete(tombstone oid.Address, toDelete []oid.ID) error {
  return e.base.Delete(tombstone, toDelete)
 }
@@ -562,6 +566,10 @@ type engineWithoutNotifications struct {
  engine *engine.StorageEngine
 }
 
+func (e engineWithoutNotifications) IsLocked(address oid.Address) (bool, error) {
+ return e.engine.IsLocked(address)
+}
+
 func (e engineWithoutNotifications) Delete(tombstone oid.Address, toDelete []oid.ID) error {
  var prm engine.InhumePrm
 

pkg/core/object/fmt.go

@@ -23,6 +23,7 @@ type FormatValidatorOption func(*cfg)
 
 type cfg struct {
  netState netmap.State
+ e LockSource
 }
 
 // DeleteHandler is an interface of delete queue processor.
@@ -34,6 +35,12 @@ type DeleteHandler interface {
  DeleteObjects(oid.Address, ...oid.Address) error
 }
 
+// LockSource is a source of lock relations between the objects.
+type LockSource interface {
+ // IsLocked must clarify object's lock status.
+ IsLocked(address oid.Address) (bool, error)
+}
+
 // Locker is an object lock storage interface.
 type Locker interface {
  // Lock list of objects as locked by locker in the specified container.
@@ -286,7 +293,24 @@ func (v *FormatValidator) checkExpiration(obj *object.Object) error {
  }
 
  if exp < v.netState.CurrentEpoch() {
- return errExpired
+ // an object could be expired but locked;
+ // put such an object is a correct operation
+
+ cID, _ := obj.ContainerID()
+ oID, _ := obj.ID()
+
+ var addr oid.Address
+ addr.SetContainer(cID)
+ addr.SetObject(oID)
+
+ locked, err := v.e.IsLocked(addr)
+ if err != nil {
+ return fmt.Errorf("locking status check for an expired object: %w", err)
+ }
+
+ if !locked {
+ return errExpired
+ }
  }
 
  return nil
@@ -347,3 +371,10 @@ func WithNetState(netState netmap.State) FormatValidatorOption {
  c.netState = netState
  }
 }
+
+// WithLockSource return option to set a Locked objects source.
+func WithLockSource(e LockSource) FormatValidatorOption {
+ return func(c *cfg) {
+ c.e = e
+ }
+}

pkg/core/object/fmt_test.go

@@ -35,13 +35,26 @@ func (s testNetState) CurrentEpoch() uint64 {
  return s.epoch
 }
 
+type testLockSource struct {
+ m map[oid.Address]bool
+}
+
+func (t testLockSource) IsLocked(address oid.Address) (bool, error) {
+ return t.m[address], nil
+}
+
 func TestFormatValidator_Validate(t *testing.T) {
  const curEpoch = 13
 
+ ls := testLockSource{
+ m: make(map[oid.Address]bool),
+ }
+
  v := NewFormatValidator(
  WithNetState(testNetState{
  epoch: curEpoch,
  }),
+ WithLockSource(ls),
  )
 
  ownerKey, err := keys.NewPrivateKey()
@@ -225,8 +238,25 @@ func TestFormatValidator_Validate(t *testing.T) {
 
  t.Run("expired object", func(t *testing.T) {
  val := strconv.FormatUint(curEpoch-1, 10)
- err := v.Validate(fn(val), false)
- require.ErrorIs(t, err, errExpired)
+ obj := fn(val)
+
+ t.Run("non-locked", func(t *testing.T) {
+ err := v.Validate(obj, false)
+ require.ErrorIs(t, err, errExpired)
+ })
+
+ t.Run("locked", func(t *testing.T) {
+ var addr oid.Address
+ oID, _ := obj.ID()
+ cID, _ := obj.ContainerID()
+
+ addr.SetContainer(cID)
+ addr.SetObject(oID)
+ ls.m[addr] = true
+
+ err := v.Validate(obj, false)
+ require.NoError(t, err)
+ })
  })
 
  t.Run("alive object", func(t *testing.T) {

pkg/local_object_storage/engine/inhume.go

@@ -81,7 +81,7 @@ func (e *StorageEngine) inhume(prm InhumePrm) (InhumeRes, error) {
 
  for i := range prm.addrs {
  if !prm.forceRemoval {
- locked, err := e.isLocked(prm.addrs[i])
+ locked, err := e.IsLocked(prm.addrs[i])
  if err != nil {
  e.log.Warn("removing an object without full locking check",
  zap.Error(err),
@@ -179,6 +179,19 @@ func (e *StorageEngine) inhumeAddr(addr oid.Address, prm shard.InhumePrm, checkE
  return ok, retErr
 }
 
+// IsLocked checks whether an object is locked according to StorageEngine's state.
+func (e *StorageEngine) IsLocked(addr oid.Address) (bool, error) {
+ var res bool
+ var err error
+
+ err = e.execIfNotBlocked(func() error {
+ res, err = e.isLocked(addr)
+ return err
+ })
+
+ return res, err
+}
+
 func (e *StorageEngine) isLocked(addr oid.Address) (bool, error) {
  var locked bool
  var err error

pkg/local_object_storage/metabase/lock.go

@@ -96,25 +96,28 @@ func (db *DB) Lock(cnr cid.ID, locker oid.ID, locked []oid.ID) error {
 }
 
 // FreeLockedBy unlocks all objects in DB which are locked by lockers.
-func (db *DB) FreeLockedBy(lockers []oid.Address) error {
+// Returns unlocked objects if any.
+func (db *DB) FreeLockedBy(lockers []oid.Address) ([]oid.Address, error) {
  db.modeMtx.RLock()
  defer db.modeMtx.RUnlock()
 
  if db.mode.NoMetabase() {
- return ErrDegradedMode
+ return nil, ErrDegradedMode
  }
 
- return db.boltDB.Update(func(tx *bbolt.Tx) error {
- var err error
+ var unlocked []oid.Address
 
+ return unlocked, db.boltDB.Update(func(tx *bbolt.Tx) error {
  for i := range lockers {
- err = freePotentialLocks(tx, lockers[i].Container(), lockers[i].Object())
+ uu, err := freePotentialLocks(tx, lockers[i].Container(), lockers[i].Object())
  if err != nil {
  return err
  }
+
+ unlocked = append(unlocked, uu...)
  }
 
- return err
+ return nil
  })
 }
 
@@ -133,60 +136,94 @@ func objectLocked(tx *bbolt.Tx, idCnr cid.ID, idObj oid.ID) bool {
  return false
 }
 
+type kv struct {
+ k []byte
+ v []byte
+}
+
 // releases all records about the objects locked by the locker.
+// Returns unlocked objects (if any).
 //
 // Operation is very resource-intensive, which is caused by the admissibility
 // of multiple locks. Also, if we knew what objects are locked, it would be
 // possible to speed up the execution.
-func freePotentialLocks(tx *bbolt.Tx, idCnr cid.ID, locker oid.ID) error {
+func freePotentialLocks(tx *bbolt.Tx, idCnr cid.ID, locker oid.ID) ([]oid.Address, error) {
  bucketLocked := tx.Bucket(bucketNameLocked)
- if bucketLocked != nil {
- key := make([]byte, cidSize)
-  idCnr.Encode(key)
+ if bucketLocked == nil {
+ return nil, nil
+ }
 
- bucketLockedContainer := bucketLocked.Bucket(key)
- if bucketLockedContainer != nil {
- keyLocker := objectKey(locker, key)
- return bucketLockedContainer.ForEach(func(k, v []byte) error {
- keyLockers, err := decodeList(v)
- if err != nil {
- return fmt.Errorf("decode list of lockers in locked bucket: %w", err)
- }
+ key := make([]byte, cidSize)
+ idCnr.Encode(key)
 
- for i := range keyLockers {
- if bytes.Equal(keyLockers[i], keyLocker) {
- if len(keyLockers) == 1 {
- // locker was all alone
- err = bucketLockedContainer.Delete(k)
- if err != nil {
- return fmt.Errorf("delete locked object record from locked bucket: %w", err)
- }
- } else {
- // exclude locker
- keyLockers = append(keyLockers[:i], keyLockers[i+1:]...)
-
- v, err = encodeList(keyLockers)
- if err != nil {
- return fmt.Errorf("encode updated list of lockers: %w", err)
- }
-
- // update the record
- err = bucketLockedContainer.Put(k, v)
- if err != nil {
- return fmt.Errorf("update list of lockers: %w", err)
- }
- }
-
- return nil
+ bucketLockedContainer := bucketLocked.Bucket(key)
+ if bucketLockedContainer == nil {
+ return nil, nil
+ }
+
+ var unlocked []oid.Address
+ var bktChanges []kv
+ keyLocker := objectKey(locker, key)
+
+ err := bucketLockedContainer.ForEach(func(k, v []byte) error {
+ keyLockers, err := decodeList(v)
+ if err != nil {
+ return fmt.Errorf("decode list of lockers in locked bucket: %w", err)
+ }
+
+ for i := range keyLockers {
+ if bytes.Equal(keyLockers[i], keyLocker) {
+ if len(keyLockers) == 1 {
+ bktChanges = append(bktChanges, kv{k: k, v: nil})
+
+ var oID oid.ID
+ err = oID.Decode(k)
+ if err != nil {
+ return fmt.Errorf("decode unlocked object id error: %w", err)
  }
+
+ var addr oid.Address
+ addr.SetContainer(idCnr)
+ addr.SetObject(oID)
+
+ unlocked = append(unlocked, addr)
+ } else {
+ // exclude locker
+ keyLockers = append(keyLockers[:i], keyLockers[i+1:]...)
+
+ v, err = encodeList(keyLockers)
+ if err != nil {
+ return fmt.Errorf("encode updated list of lockers: %w", err)
+ }
+
+ bktChanges = append(bktChanges, kv{k: k, v: v})
  }
 
  return nil
- })
+ }
+ }
+
+ return nil
+ })
+ if err != nil {
+ return nil, fmt.Errorf("iterating lockers: %w", err)
+ }
+
+ for _, kv := range bktChanges {
+ if kv.v == nil {
+ err = bucketLockedContainer.Delete(kv.k)
+ if err != nil {
+ return nil, fmt.Errorf("delete locked object record from locked bucket: %w", err)
+ }
+ } else {
+ err = bucketLockedContainer.Put(kv.k, kv.v)
+ if err != nil {
+ return nil, fmt.Errorf("update list of lockers: %w", err)
+ }
  }
  }
 
- return nil
+ return unlocked, nil
 }
 
 // IsLockedPrm groups the parameters of IsLocked operation.

pkg/local_object_storage/metabase/lock_test.go

@@ -109,8 +109,9 @@ func TestDB_Lock(t *testing.T) {
  require.Len(t, res.DeletedLockObjects(), 1)
  require.Equal(t, objectcore.AddressOf(lockObj), res.DeletedLockObjects()[0])
 
- err = db.FreeLockedBy([]oid.Address{lockAddr})
+ unlocked, err := db.FreeLockedBy([]oid.Address{lockAddr})
  require.NoError(t, err)
+ require.ElementsMatch(t, objsToAddrs(objs), unlocked)
 
  inhumePrm.SetAddresses(objAddr)
  inhumePrm.SetGCMark()
@@ -140,8 +141,9 @@ func TestDB_Lock(t *testing.T) {
 
  // unlock just objects that were locked by
  // just removed locker
- err = db.FreeLockedBy([]oid.Address{res.DeletedLockObjects()[0]})
+ unlocked, err := db.FreeLockedBy([]oid.Address{res.DeletedLockObjects()[0]})
  require.NoError(t, err)
+ require.ElementsMatch(t, objsToAddrs(objs), unlocked)
 
  // removing objects after unlock
 
@@ -264,3 +266,12 @@ func putAndLockObj(t *testing.T, db *meta.DB, numOfLockedObjs int) ([]*object.Ob
 
  return lockedObjs, lockObj
 }
+
+func objsToAddrs(oo []*object.Object) []oid.Address {
+ res := make([]oid.Address, 0, len(oo))
+ for _, o := range oo {
+ res = append(res, objectcore.AddressOf(o))
+ }
+
+ return res
+}