From 337049b2ce53884344b97c0be1a222d38bdc72f9 Mon Sep 17 00:00:00 2001 From: Pavel Karpy Date: Mon, 30 Jan 2023 13:23:44 +0300 Subject: [PATCH] [#56] node: Allow reading expired locked object Signed-off-by: Pavel Karpy --- CHANGELOG.md | 1 + pkg/local_object_storage/metabase/exists.go | 5 +++++ .../metabase/lock_test.go | 21 +++++++++++++++++++ 3 files changed, 27 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b448e265..c03a2654b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ Changelog for FrostFS Node - Pilorama now can merge multiple batches into one (#2231) - Storage engine now can start even when some shard components are unavailable (#2238) - `neofs-cli` buffer for object put increased from 4 KiB to 3 MiB (#2243) +- Expired locked object is available for reading (#56) ### Fixed - Increase payload size metric on shards' `put` operation (#1794) diff --git a/pkg/local_object_storage/metabase/exists.go b/pkg/local_object_storage/metabase/exists.go index 3a021288d..d9b1ed520 100644 --- a/pkg/local_object_storage/metabase/exists.go +++ b/pkg/local_object_storage/metabase/exists.go @@ -100,6 +100,11 @@ func (db *DB) exists(tx *bbolt.Tx, addr oid.Address, currEpoch uint64) (exists b // - 2 if object is covered with tombstone; // - 3 if object is expired. func objectStatus(tx *bbolt.Tx, addr oid.Address, currEpoch uint64) uint8 { + // locked object could not be removed/marked with GC/expired + if objectLocked(tx, addr.Container(), addr.Object()) { + return 0 + } + // we check only if the object is expired in the current // epoch since it is considered the only corner case: the // GC is expected to collect all the objects that have diff --git a/pkg/local_object_storage/metabase/lock_test.go b/pkg/local_object_storage/metabase/lock_test.go index 9ae8ec0da..d7c6a153f 100644 --- a/pkg/local_object_storage/metabase/lock_test.go +++ b/pkg/local_object_storage/metabase/lock_test.go @@ -169,6 +169,27 @@ func TestDB_Lock(t *testing.T) { }) } +func TestDB_Lock_Expired(t *testing.T) { + es := &epochState{e: 123} + + db := newDB(t, meta.WithEpochState(es)) + + // put an object + addr := putWithExpiration(t, db, object.TypeRegular, 124) + + // expire the obj + es.e = 125 + _, err := metaGet(db, addr, false) + require.ErrorIs(t, err, meta.ErrObjectIsExpired) + + // lock the obj + require.NoError(t, db.Lock(addr.Container(), oidtest.ID(), []oid.ID{addr.Object()})) + + // object is expired but locked, thus, must be available + _, err = metaGet(db, addr, false) + require.NoError(t, err) +} + func TestDB_IsLocked(t *testing.T) { db := newDB(t)