From 9a039ba5825ccca4f73653741d1fe61ffa397c75 Mon Sep 17 00:00:00 2001
From: Pavel Karpy <carpawell@nspcc.ru>
Date: Sat, 12 Nov 2022 14:46:44 +0300
Subject: [PATCH] [#1502] meta: Add `IsLocked` method

It gets an object and returns its locking status.

Signed-off-by: Pavel Karpy <carpawell@nspcc.ru>
---
 pkg/local_object_storage/metabase/lock.go     | 32 ++++++++++++++
 .../metabase/lock_test.go                     | 44 +++++++++++++++++++
 2 files changed, 76 insertions(+)

diff --git a/pkg/local_object_storage/metabase/lock.go b/pkg/local_object_storage/metabase/lock.go
index 786b3d7c..a15a9b6f 100644
--- a/pkg/local_object_storage/metabase/lock.go
+++ b/pkg/local_object_storage/metabase/lock.go
@@ -175,3 +175,35 @@ func freePotentialLocks(tx *bbolt.Tx, idCnr cid.ID, locker oid.ID) error {
 
 	return nil
 }
+
+// IsLockedPrm groups the parameters of IsLocked operation.
+type IsLockedPrm struct {
+	addr oid.Address
+}
+
+// SetAddress sets object address that will be checked for lock relations.
+func (i *IsLockedPrm) SetAddress(addr oid.Address) {
+	i.addr = addr
+}
+
+// IsLockedRes groups the resulting values of IsLocked operation.
+type IsLockedRes struct {
+	locked bool
+}
+
+// Locked describes the requested object status according to the metabase
+// current state.
+func (i IsLockedRes) Locked() bool {
+	return i.locked
+}
+
+// IsLocked checks is the provided object is locked by any `LOCK`. Not found
+// object is considered as non-locked.
+//
+// Returns only non-logical errors related to underlying database.
+func (db *DB) IsLocked(prm IsLockedPrm) (res IsLockedRes, err error) {
+	return res, db.boltDB.View(func(tx *bbolt.Tx) error {
+		res.locked = objectLocked(tx, prm.addr.Container(), prm.addr.Object())
+		return nil
+	})
+}
diff --git a/pkg/local_object_storage/metabase/lock_test.go b/pkg/local_object_storage/metabase/lock_test.go
index 1bd16754..e06b7039 100644
--- a/pkg/local_object_storage/metabase/lock_test.go
+++ b/pkg/local_object_storage/metabase/lock_test.go
@@ -169,6 +169,50 @@ func TestDB_Lock(t *testing.T) {
 	})
 }
 
+func TestDB_IsLocked(t *testing.T) {
+	db := newDB(t)
+
+	// existing and locked objs
+
+	objs, _ := putAndLockObj(t, db, 5)
+	var prm meta.IsLockedPrm
+
+	for _, obj := range objs {
+		prm.SetAddress(objectcore.AddressOf(obj))
+
+		res, err := db.IsLocked(prm)
+		require.NoError(t, err)
+
+		require.True(t, res.Locked())
+	}
+
+	// some rand obj
+
+	prm.SetAddress(oidtest.Address())
+
+	res, err := db.IsLocked(prm)
+	require.NoError(t, err)
+
+	require.False(t, res.Locked())
+
+	// existing but not locked obj
+
+	obj := objecttest.Object()
+
+	var putPrm meta.PutPrm
+	putPrm.SetObject(obj)
+
+	_, err = db.Put(putPrm)
+	require.NoError(t, err)
+
+	prm.SetAddress(objectcore.AddressOf(obj))
+
+	res, err = db.IsLocked(prm)
+	require.NoError(t, err)
+
+	require.False(t, res.Locked())
+}
+
 // putAndLockObj puts object, returns it and its locker.
 func putAndLockObj(t *testing.T, db *meta.DB, numOfLockedObjs int) ([]*object.Object, *object.Object) {
 	cnr := cidtest.ID()