meta: Do not change bucket while performing ForEach

See boltdb/bolt#428.

Signed-off-by: Pavel Karpy <[email protected]>

pkg/local_object_storage/metabase/lock.go

@@ -136,6 +136,11 @@ 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).
 //
@@ -157,9 +162,10 @@ func freePotentialLocks(tx *bbolt.Tx, idCnr cid.ID, locker oid.ID) ([]oid.Addres
  }
 
  var unlocked []oid.Address
+ var bktChanges []kv
  keyLocker := objectKey(locker, key)
 
- return unlocked, bucketLockedContainer.ForEach(func(k, v []byte) error {
+ 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)
@@ -168,11 +174,7 @@ func freePotentialLocks(tx *bbolt.Tx, idCnr cid.ID, locker oid.ID) ([]oid.Addres
  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)
- }
+ bktChanges = append(bktChanges, kv{k: k, v: nil})
 
  var oID oid.ID
  err = oID.Decode(k)
@@ -194,11 +196,7 @@ func freePotentialLocks(tx *bbolt.Tx, idCnr cid.ID, locker oid.ID) ([]oid.Addres
  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)
- }
+ bktChanges = append(bktChanges, kv{k: k, v: v})
  }
 
  return nil
@@ -207,6 +205,25 @@ func freePotentialLocks(tx *bbolt.Tx, idCnr cid.ID, locker oid.ID) ([]oid.Addres
 
  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 unlocked, nil
 }
 
 // IsLockedPrm groups the parameters of IsLocked operation.