diff --git a/pkg/local_object_storage/metabase/iterators.go b/pkg/local_object_storage/metabase/iterators.go index 3d463e49c..64a2c0eb0 100644 --- a/pkg/local_object_storage/metabase/iterators.go +++ b/pkg/local_object_storage/metabase/iterators.go @@ -88,7 +88,7 @@ func (db *DB) iterateExpired(tx *bbolt.Tx, epoch uint64, h ExpiredObjectHandler) addr.SetObjectID(id) return h(&ExpiredObject{ - typ: objectType(tx, cnrID, idKey), + typ: firstIrregularObjectType(tx, *cnrID, idKey), addr: addr, }) }) @@ -102,17 +102,6 @@ func (db *DB) iterateExpired(tx *bbolt.Tx, epoch uint64, h ExpiredObjectHandler) return err } -func objectType(tx *bbolt.Tx, cid *cid.ID, oidBytes []byte) object.Type { - switch { - default: - return object.TypeRegular - case inBucket(tx, tombstoneBucketName(cid), oidBytes): - return object.TypeTombstone - case inBucket(tx, storageGroupBucketName(cid), oidBytes): - return object.TypeStorageGroup - } -} - // IterateCoveredByTombstones iterates over all objects in DB which are covered // by tombstone with string address from tss. // diff --git a/pkg/local_object_storage/metabase/lock.go b/pkg/local_object_storage/metabase/lock.go index 5e5df53f4..eb22e4a20 100644 --- a/pkg/local_object_storage/metabase/lock.go +++ b/pkg/local_object_storage/metabase/lock.go @@ -1,13 +1,99 @@ package meta import ( + "bytes" + "errors" + "fmt" + cid "github.com/nspcc-dev/neofs-sdk-go/container/id" + "github.com/nspcc-dev/neofs-sdk-go/object" + oid "github.com/nspcc-dev/neofs-sdk-go/object/id" + "go.etcd.io/bbolt" ) -// suffix for container buckets with locked objects. -const bucketNameSuffixLocked = invalidBase58String + "LOCKED" +// bucket name for locked objects. +var bucketNameLocked = []byte(invalidBase58String + "Locked") -// returns name of the bucket with locked objects for specified container. -func bucketNameLocked(idCnr cid.ID) []byte { - return []byte(idCnr.String() + bucketNameSuffixLocked) +// suffix for container buckets with objects of type LOCK. +const bucketNameSuffixLockers = invalidBase58String + "LOCKER" + +// returns name of the bucket with objects of type LOCK for specified container. +func bucketNameLockers(idCnr cid.ID) []byte { + return []byte(idCnr.String() + bucketNameSuffixLockers) +} + +// ErrLockIrregularObject is returned when trying to lock an irregular object. +var ErrLockIrregularObject = errors.New("locking irregular object") + +// Lock marks objects as locked with another object. All objects are from the +// specified container. +// +// Allows locking regular objects only (otherwise returns ErrLockIrregularObject). +// +// Locked list should be unique. Panics if it is empty. +func (db *DB) Lock(cnr cid.ID, locker oid.ID, locked []oid.ID) error { + if len(locked) == 0 { + panic("empty locked list") + } + + return db.boltDB.Update(func(tx *bbolt.Tx) error { + // check if all objects are regular + bucketKeysLocked := make([][]byte, len(locked)) + + for i := range locked { + bucketKeysLocked[i] = objectKey(&locked[i]) + } + + if firstIrregularObjectType(tx, cnr, bucketKeysLocked...) != object.TypeRegular { + return ErrLockIrregularObject + } + + bucketLocked, err := tx.CreateBucketIfNotExists(bucketNameLocked) + if err != nil { + return fmt.Errorf("create global bucket for locked objects: %w", err) + } + + bucketLockedContainer, err := bucketLocked.CreateBucketIfNotExists([]byte(cnr.String())) + if err != nil { + return fmt.Errorf("create container bucket for locked objects %v: %w", cnr, err) + } + + keyLocker := objectKey(&locker) + var exLockers [][]byte + var updLockers []byte + + loop: + for i := range bucketKeysLocked { + // decode list of already existing lockers + exLockers, err = decodeList(bucketLockedContainer.Get(bucketKeysLocked[i])) + if err != nil { + return fmt.Errorf("decode list of object lockers: %w", err) + } + + for i := range exLockers { + if bytes.Equal(exLockers[i], keyLocker) { + continue loop + } + } + + // update the list of lockers + if exLockers == nil { + updLockers = keyLocker + } else { + updLockers, err = encodeList(append(exLockers, keyLocker)) + if err != nil { + // maybe continue for the best effort? + return fmt.Errorf("encode list of object lockers: %w", err) + } + } + + // write updated list of lockers + err = bucketLockedContainer.Put(bucketKeysLocked[i], updLockers) + if err != nil { + return fmt.Errorf("update list of object lockers: %w", err) + } + } + + return nil + }) } diff --git a/pkg/local_object_storage/metabase/lock_test.go b/pkg/local_object_storage/metabase/lock_test.go new file mode 100644 index 000000000..ca567fcd4 --- /dev/null +++ b/pkg/local_object_storage/metabase/lock_test.go @@ -0,0 +1,49 @@ +package meta_test + +import ( + "testing" + + objectCore "github.com/nspcc-dev/neofs-node/pkg/core/object" + meta "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/metabase" + cidtest "github.com/nspcc-dev/neofs-sdk-go/container/id/test" + "github.com/nspcc-dev/neofs-sdk-go/object" + oid "github.com/nspcc-dev/neofs-sdk-go/object/id" + oidtest "github.com/nspcc-dev/neofs-sdk-go/object/id/test" + objecttest "github.com/nspcc-dev/neofs-sdk-go/object/test" + "github.com/stretchr/testify/require" +) + +func TestDB_Lock(t *testing.T) { + cnr := *cidtest.ID() + db := newDB(t) + + t.Run("empty locked list", func(t *testing.T) { + require.Panics(t, func() { _ = db.Lock(cnr, oid.ID{}, nil) }) + require.Panics(t, func() { _ = db.Lock(cnr, oid.ID{}, []oid.ID{}) }) + }) + + t.Run("(ir)regular", func(t *testing.T) { + for _, typ := range [...]object.Type{ + object.TypeTombstone, + object.TypeStorageGroup, + object.TypeLock, + object.TypeRegular, + } { + obj := objecttest.Raw() + obj.SetType(typ) + obj.SetContainerID(&cnr) + + // save irregular object + err := meta.Put(db, objectCore.NewFromSDK(obj.Object()), nil) + require.NoError(t, err, typ) + + // try to lock it + err = db.Lock(cnr, *oidtest.ID(), []oid.ID{*obj.ID()}) + if typ == object.TypeRegular { + require.NoError(t, err, typ) + } else { + require.ErrorIs(t, err, meta.ErrLockIrregularObject, typ) + } + } + }) +} diff --git a/pkg/local_object_storage/metabase/util.go b/pkg/local_object_storage/metabase/util.go index 7a6c58a81..dfbec719b 100644 --- a/pkg/local_object_storage/metabase/util.go +++ b/pkg/local_object_storage/metabase/util.go @@ -132,3 +132,31 @@ func resetBucket(b *bbolt.Bucket) error { return b.DeleteBucket(k) }) } + +// if meets irregular object container in objs - returns its type, otherwise returns object.TypeRegular. +// +// firstIrregularObjectType(tx, cnr, obj) usage allows getting object type. +func firstIrregularObjectType(tx *bbolt.Tx, idCnr cid.ID, objs ...[]byte) object.Type { + if len(objs) == 0 { + panic("empty object list in firstIrregularObjectType") + } + + irregularTypeBuckets := [...]struct { + typ object.Type + name []byte + }{ + {object.TypeTombstone, tombstoneBucketName(&idCnr)}, + {object.TypeStorageGroup, storageGroupBucketName(&idCnr)}, + {object.TypeLock, bucketNameLockers(idCnr)}, + } + + for i := range objs { + for j := range irregularTypeBuckets { + if inBucket(tx, irregularTypeBuckets[j].name, objs[i]) { + return irregularTypeBuckets[j].typ + } + } + } + + return object.TypeRegular +}